Day 33: Vector DB Production

user query
  -> normalize + embed query
  -> vector search top_k
  -> metadata/ACL filtering
  -> optional BM25/hybrid search
  -> rerank
  -> build context with citations
  -> LLM answer

score_i = similarity(query_vector, document_vector_i)
sort score desc
return top_k

{
  "id": "company_a:policy_001:2026-01:chunk_00012",
  "document_id": "policy_001",
  "chunk_id": "chunk_00012",
  "text": "Nhân viên full-time có 12 ngày nghỉ phép năm...",
  "vector": [0.01, -0.04, 0.21],
  "metadata": {
    "tenant_id": "company_a",
    "acl_roles": ["employee", "hr"],
    "source_uri": "s3://kb/company_a/hr/policy.pdf",
    "source_type": "pdf",
    "document_version": "2026-01",
    "chunk_index": 12,
    "page_start": 3,
    "page_end": 4,
    "section_path": ["HR", "Leave Policy"],
    "language": "vi",
    "embedding_model": "BAAI/bge-m3",
    "dimension": 1024,
    "metric": "cosine",
    "chunking_strategy": "markdown_heading_v2_800_120",
    "index_version": "rag-index-2026-05-10-bge-m3-v2",
    "text_hash": "sha256:...",
    "deleted_at": null,
    "created_at": "2026-05-10T08:00:00Z",
    "updated_at": "2026-05-10T08:00:00Z"
  }
}

tenant_id = current_user.tenant_id
AND deleted_at IS NULL
AND acl_roles intersects current_user.roles
AND index_version = active_index_version

from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Sequence

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance,
    FieldCondition,
    Filter,
    HnswConfigDiff,
    MatchAny,
    MatchValue,
    PayloadSchemaType,
    PointStruct,
    VectorParams,
)


COLLECTION = "rag_chunks_bge_m3_v2"
VECTOR_SIZE = 1024
ACTIVE_INDEX_VERSION = "rag-index-2026-05-10-bge-m3-v2"


@dataclass(frozen=True)
class AuthContext:
    tenant_id: str
    roles: tuple[str, ...]


def get_client() -> QdrantClient:
    return QdrantClient(
        url=os.environ["QDRANT_URL"],
        api_key=os.environ.get("QDRANT_API_KEY"),
        timeout=10,
    )


def ensure_collection(client: QdrantClient) -> None:
    existing = {c.name for c in client.get_collections().collections}
    if COLLECTION not in existing:
        client.create_collection(
            collection_name=COLLECTION,
            vectors_config=VectorParams(size=VECTOR_SIZE, distance=Distance.COSINE),
            hnsw_config=HnswConfigDiff(m=16, ef_construct=128),
            on_disk_payload=True,
        )

    for field in ("tenant_id", "acl_roles", "document_id", "index_version", "deleted"):
        client.create_payload_index(
            collection_name=COLLECTION,
            field_name=field,
            field_schema=PayloadSchemaType.KEYWORD,
        )


def upsert_chunks(client: QdrantClient, chunks: Sequence[dict]) -> None:
    points = []
    for chunk in chunks:
        payload = {
            **chunk["metadata"],
            "text": chunk["text"],
            "deleted": "false",
            "index_version": ACTIVE_INDEX_VERSION,
        }
        points.append(
            PointStruct(
                id=chunk["id"],
                vector=chunk["embedding"],
                payload=payload,
            )
        )

    client.upsert(collection_name=COLLECTION, points=points, wait=True)


def search_chunks(
    client: QdrantClient,
    query_vector: list[float],
    auth: AuthContext,
    limit: int = 20,
):
    query_filter = Filter(
        must=[
            FieldCondition(key="tenant_id", match=MatchValue(value=auth.tenant_id)),
            FieldCondition(key="deleted", match=MatchValue(value="false")),
            FieldCondition(key="index_version", match=MatchValue(value=ACTIVE_INDEX_VERSION)),
            FieldCondition(key="acl_roles", match=MatchAny(any=list(auth.roles))),
        ]
    )

    return client.query_points(
        collection_name=COLLECTION,
        query=query_vector,
        query_filter=query_filter,
        limit=limit,
        with_payload=[
            "text",
            "document_id",
            "chunk_id",
            "source_uri",
            "page_start",
            "page_end",
            "section_path",
            "index_version",
        ],
        with_vectors=False,
    ).points

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE rag_chunks (
    id TEXT PRIMARY KEY,
    tenant_id TEXT NOT NULL,
    document_id TEXT NOT NULL,
    chunk_id TEXT NOT NULL,
    text TEXT NOT NULL,
    embedding vector(1024) NOT NULL,
    acl_roles TEXT[] NOT NULL,
    source_uri TEXT NOT NULL,
    page_start INTEGER,
    page_end INTEGER,
    embedding_model TEXT NOT NULL,
    metric TEXT NOT NULL DEFAULT 'cosine',
    chunking_strategy TEXT NOT NULL,
    index_version TEXT NOT NULL,
    deleted_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX rag_chunks_tenant_idx
    ON rag_chunks (tenant_id, index_version)
    WHERE deleted_at IS NULL;

CREATE INDEX rag_chunks_document_idx
    ON rag_chunks (tenant_id, document_id);

CREATE INDEX rag_chunks_acl_roles_idx
    ON rag_chunks USING gin (acl_roles);

CREATE INDEX CONCURRENTLY rag_chunks_embedding_hnsw_idx
    ON rag_chunks USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);

BEGIN;
SET LOCAL hnsw.ef_search = 100;

SELECT
    id,
    document_id,
    chunk_id,
    text,
    source_uri,
    page_start,
    page_end,
    1 - (embedding <=> $1::vector) AS similarity
FROM rag_chunks
WHERE tenant_id = $2
  AND deleted_at IS NULL
  AND index_version = $3
  AND acl_roles && $4::text[]
ORDER BY embedding <=> $1::vector
LIMIT $5;

COMMIT;

current index: rag-index-2026-05-01-bge-m3-v1
new index:     rag-index-2026-05-10-bge-m3-v2

1. Build new collection/index ở background.
2. Ingest toàn bộ documents sang index mới.
3. Chạy eval: Recall@K, MRR, citation accuracy, latency, ACL tests.
4. Chạy shadow traffic nếu có.
5. Switch active_index_version.
6. Giữ index cũ đủ lâu để rollback.
7. Xóa index cũ sau retention window.

{
  "request_id": "req_123",
  "tenant_id": "company_a",
  "user_id_hash": "u_hash",
  "index_version": "rag-index-2026-05-10-bge-m3-v2",
  "embedding_model": "BAAI/bge-m3",
  "top_k": 20,
  "filters": ["tenant_id", "acl_roles", "deleted", "index_version"],
  "result_chunk_ids": ["chunk_1", "chunk_2"],
  "scores": [0.82, 0.79],
  "latency_ms": {
    "embed": 45,
    "vector_search": 18,
    "rerank": 70
  }
}

Documents
  -> parse
  -> chunk
  -> embed
  -> upsert vector + metadata
  -> query vector + mandatory filters
  -> rerank
  -> answer with citations

raw_vector_bytes = number_of_chunks * dimension * bytes_per_float

| Config | Recall@5 | MRR@10 | p95 search ms | RAM/storage | Ghi chú |
|---|---:|---:|---:|---:|---|
| exact baseline | | | | | |
| hnsw default | | | | | |
| hnsw ef_search=100 | | | | | |
| hnsw ef_search=200 | | | | | |
| quantized | | | | | |

services:
  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - qdrant_data:/qdrant/storage
    environment:
      QDRANT__SERVICE__API_KEY: "dev-local-key-change-me"

volumes:
  qdrant_data:

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE rag_chunks (
    id TEXT PRIMARY KEY,
    tenant_id TEXT NOT NULL,
    document_id TEXT NOT NULL,
    chunk_id TEXT NOT NULL,
    text TEXT NOT NULL,
    embedding vector(1024) NOT NULL,
    metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
    acl_roles TEXT[] NOT NULL,
    index_version TEXT NOT NULL,
    deleted_at TIMESTAMPTZ
);

CREATE INDEX rag_chunks_filter_idx
    ON rag_chunks (tenant_id, index_version)
    WHERE deleted_at IS NULL;

CREATE INDEX rag_chunks_acl_idx
    ON rag_chunks USING gin (acl_roles);

CREATE INDEX CONCURRENTLY rag_chunks_hnsw_idx
    ON rag_chunks USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);

tenant_id + document_id + document_version + chunking_strategy + embedding_model + chunk_index

services:
  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
    volumes:
      - qdrant_data:/qdrant/storage

volumes:
  qdrant_data:

docker compose up -d

CHUNKS = [
    {
        "id": "a:hr:leave:001",
        "text": "Nhân viên công ty A có 12 ngày nghỉ phép năm.",
        "vector": [0.90, 0.10, 0.00, 0.00],
        "metadata": {
            "tenant_id": "company_a",
            "document_id": "hr_leave",
            "chunk_id": "001",
            "acl_roles": ["employee", "hr"],
            "source_uri": "s3://company-a/hr/leave.pdf",
            "page_start": 1,
            "page_end": 1,
            "index_version": "dev-index-v1",
            "deleted": "false",
        },
    },
    {
        "id": "a:finance:salary:001",
        "text": "Bảng lương chi tiết chỉ dành cho phòng finance.",
        "vector": [0.85, 0.15, 0.00, 0.05],
        "metadata": {
            "tenant_id": "company_a",
            "document_id": "finance_salary",
            "chunk_id": "001",
            "acl_roles": ["finance"],
            "source_uri": "s3://company-a/finance/salary.pdf",
            "page_start": 2,
            "page_end": 2,
            "index_version": "dev-index-v1",
            "deleted": "false",
        },
    },
    {
        "id": "b:hr:leave:001",
        "text": "Nhân viên công ty B có 15 ngày nghỉ phép năm.",
        "vector": [0.91, 0.09, 0.00, 0.00],
        "metadata": {
            "tenant_id": "company_b",
            "document_id": "hr_leave",
            "chunk_id": "001",
            "acl_roles": ["employee", "hr"],
            "source_uri": "s3://company-b/hr/leave.pdf",
            "page_start": 1,
            "page_end": 1,
            "index_version": "dev-index-v1",
            "deleted": "false",
        },
    },
]

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance,
    FieldCondition,
    Filter,
    HnswConfigDiff,
    MatchAny,
    MatchValue,
    PayloadSchemaType,
    PointStruct,
    VectorParams,
)

COLLECTION = "day33_chunks"
INDEX_VERSION = "dev-index-v1"

client = QdrantClient(url="http://localhost:6333")

if COLLECTION not in {c.name for c in client.get_collections().collections}:
    client.create_collection(
        collection_name=COLLECTION,
        vectors_config=VectorParams(size=4, distance=Distance.COSINE),
        hnsw_config=HnswConfigDiff(m=16, ef_construct=100),
        on_disk_payload=True,
    )

for field in ["tenant_id", "acl_roles", "document_id", "index_version", "deleted"]:
    client.create_payload_index(
        collection_name=COLLECTION,
        field_name=field,
        field_schema=PayloadSchemaType.KEYWORD,
    )

def upsert_chunks(chunks: list[dict]) -> None:
    points = [
        PointStruct(
            id=item["id"],
            vector=item["vector"],
            payload={**item["metadata"], "text": item["text"]},
        )
        for item in chunks
    ]
    client.upsert(collection_name=COLLECTION, points=points, wait=True)

def search(query_vector: list[float], tenant_id: str, roles: list[str], limit: int = 5):
    query_filter = Filter(
        must=[
            FieldCondition(key="tenant_id", match=MatchValue(value=tenant_id)),
            FieldCondition(key="index_version", match=MatchValue(value=INDEX_VERSION)),
            FieldCondition(key="deleted", match=MatchValue(value="false")),
            FieldCondition(key="acl_roles", match=MatchAny(any=roles)),
        ]
    )

    response = client.query_points(
        collection_name=COLLECTION,
        query=query_vector,
        query_filter=query_filter,
        limit=limit,
        with_payload=["text", "tenant_id", "document_id", "acl_roles", "source_uri"],
        with_vectors=False,
    )
    return response.points

results = search([0.90, 0.10, 0.00, 0.00], tenant_id="company_a", roles=["employee"])
for point in results:
    print(point.id, point.score, point.payload)

def test_company_a_employee_cannot_see_company_b():
    results = search([0.90, 0.10, 0.00, 0.00], "company_a", ["employee"], limit=20)
    assert results
    assert all(point.payload["tenant_id"] == "company_a" for point in results)


def test_employee_cannot_see_finance_only_document():
    results = search([0.85, 0.15, 0.00, 0.05], "company_a", ["employee"], limit=20)
    document_ids = {point.payload["document_id"] for point in results}
    assert "finance_salary" not in document_ids


def test_finance_can_see_finance_document():
    results = search([0.85, 0.15, 0.00, 0.05], "company_a", ["finance"], limit=20)
    document_ids = {point.payload["document_id"] for point in results}
    assert "finance_salary" in document_ids

import statistics
import time


QUERIES = [
    ([0.90, 0.10, 0.00, 0.00], "company_a", ["employee"]),
    ([0.85, 0.15, 0.00, 0.05], "company_a", ["finance"]),
    ([0.91, 0.09, 0.00, 0.00], "company_b", ["employee"]),
]


def percentile(values: list[float], p: int) -> float:
    values = sorted(values)
    index = int((len(values) - 1) * p / 100)
    return values[index]


latencies_ms = []
for _ in range(100):
    for query_vector, tenant_id, roles in QUERIES:
        started = time.perf_counter()
        search(query_vector, tenant_id, roles)
        latencies_ms.append((time.perf_counter() - started) * 1000)

print("p50_ms", statistics.median(latencies_ms))
print("p95_ms", percentile(latencies_ms, 95))

QRELS = [
    {
        "query_vector": [0.90, 0.10, 0.00, 0.00],
        "tenant_id": "company_a",
        "roles": ["employee"],
        "expected_document_id": "hr_leave",
    },
    {
        "query_vector": [0.91, 0.09, 0.00, 0.00],
        "tenant_id": "company_b",
        "roles": ["employee"],
        "expected_document_id": "hr_leave",
    },
]


def hit_at_k(k: int = 5) -> float:
    hits = 0
    for item in QRELS:
        results = search(item["query_vector"], item["tenant_id"], item["roles"], limit=k)
        document_ids = [point.payload["document_id"] for point in results]
        if item["expected_document_id"] in document_ids:
            hits += 1
    return hits / len(QRELS)


print("Hit@5", hit_at_k(5))

client.set_payload(
    collection_name=COLLECTION,
    payload={"deleted": "true"},
    points=["a:finance:salary:001"],
    wait=True,
)

results = search([0.85, 0.15, 0.00, 0.05], "company_a", ["finance"], limit=20)
assert "finance_salary" not in {point.payload["document_id"] for point in results}

# Day 33 Vector DB Benchmark Report

## Decision

- Chọn Vector DB:
- Lý do:
- Khi nào cần đổi lựa chọn:

## Schema

- Vector dimension:
- Metric:
- Metadata bắt buộc:
- Tenant/ACL strategy:

## Benchmark

| Config | Hit@5 | p50 ms | p95 ms | Notes |
|---|---:|---:|---:|---|
| qdrant hnsw default | | | | |

## Security Tests

- company_a không thấy company_b:
- employee không thấy finance:
- deleted document không được retrieve:

## Production Readiness

- Backup plan:
- Reindex plan:
- Delete/update plan:
- Monitoring:
- Rủi ro còn lại: