add vector store

.gitignore

@@ -2,6 +2,9 @@ __pycache__/
 *.py[cod]
 *$py.class
 postman_collection.json
+
+# Runtime data (vector stores, SQLite DBs, model caches, etc.)
+data/
 *.so
 
 .Python
@@ -121,5 +124,5 @@ logs/
 *.temp
 
 local_deploy.py
-
+test_vector_store_async.py
 deploy_sdk.py

app/api/deps.py

@@ -11,6 +11,7 @@ from app.services.extraction_service import ExtractionService
 from app.services.ocr_service import OCRService
 from app.services.sql_validator_service import SqlValidatorService
 from app.services.text_cleaner_service import TextCleanerService
+from app.services.vector_store_service import VectorStoreService
 from app.services.web_search_service import WebSearchService
 
 
@@ -47,5 +48,10 @@ def get_embeddings_service() -> EmbeddingService:
     return _embedding_service
 
 
+def get_vector_store_service() -> VectorStoreService:
+    from app.api.server import _vector_store_service
+    return _vector_store_service
+
+
 def require_auth(token: str = Depends(require_api_key)) -> str:
     return token

app/api/server.py

@@ -13,13 +13,16 @@ from app.core.database import pool_manager
 from app.core.logger import get_logger
 from app.core.redis_client import create_redis_client, close_redis
 from app.core.scripts import load_scripts
+from app.core.vector_store.deps import init_vector_store_db
 from app.services.embeddings_service import EmbeddingService
+from app.services.vector_store_service import VectorStoreService
 from app.api.v1.router import api_v1_router
 
 _logger = get_logger(__name__)
 _settings = get_settings()
 
 _embedding_service: EmbeddingService = EmbeddingService()
+_vector_store_service: VectorStoreService = VectorStoreService(_embedding_service)
 
 
 async def _self_ping():
@@ -44,11 +47,17 @@ async def lifespan(app: FastAPI):
     await init_auth_db()
     _logger.info("Authentication database initialized")
 
+    _logger.info("Initializing vector store database...")
+    await init_vector_store_db()
+    await _vector_store_service.init_db()
+    _logger.info("Vector store database initialized with %d stores", len(_vector_store_service.list_stores()))
+
     _logger.info("Initializing embedding service (loading 384-dim model)...")
     loop = asyncio.get_running_loop()
     await loop.run_in_executor(None, _embedding_service.load_model, 384)
     # await loop.run_in_executor(None, _embedding_service.load_vision_model)  # DISABLED (OOM mitigation)
     _logger.info("Embedding service initialized with dims: %s", _embedding_service.loaded_dimensions)
+    _logger.info("Vector store service initialized with %d existing stores", len(_vector_store_service.list_stores()))
 
     redis = create_redis_client(_settings.redis_url) if _settings.redis_url else None
     scripts = await load_scripts(redis) if redis else {}
@@ -63,6 +72,7 @@ async def lifespan(app: FastAPI):
     yield
     _logger.info("Shutting down...")
     await close_redis(redis)
+    await _vector_store_service.close_all()
     await pool_manager.close_all()
 
 
@@ -79,6 +89,7 @@ def create_application() -> FastAPI:
             {"name": "System", "description": "Health, info, and supported formats"},
             {"name": "Embeddings", "description": "Text embedding generation using transformer models"},
             {"name": "Verify", "description": "Phone number and identity verification"},
+            {"name": "Vector Stores", "description": "Create, manage, and search vector stores for RAG"},
         ],
         lifespan=lifespan,
     )
@@ -124,7 +135,17 @@ def create_application() -> FastAPI:
 
     @app.get("/health", include_in_schema=False)
     async def root_health():
-        return {"status": "ok", "version": _settings.app_version}
+        store_count = len(_vector_store_service.list_stores())
+        doc_count = await _vector_store_service.get_total_document_count()
+        return {
+            "success": True,
+            "app_name": _settings.app_name,
+            "version": _settings.app_version,
+            "embedding_dimension": _settings.embedding_dimension,
+            "vector_store_count": store_count,
+            "total_documents": doc_count,
+            "model_loaded": _embedding_service.is_loaded(384),
+        }
 
     @app.get("/ping", include_in_schema=False)
     async def ping():

app/api/v1/router.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from fastapi import APIRouter
 
-from app.api.v1 import auth, batch, chat, code_executor, convert, database, embeddings, reconcile, scraper, semantic_router, sql_validator, system, token_counter, token_generator, web_search
+from app.api.v1 import auth, batch, chat, code_executor, convert, database, embeddings, reconcile, scraper, semantic_router, sql_validator, system, token_counter, token_generator, vector_stores, web_search
 from app.api.verify import router as verify_router
 
 api_v1_router = APIRouter()
@@ -22,3 +22,4 @@ api_v1_router.include_router(semantic_router.router, tags=["Semantic Router"])
 api_v1_router.include_router(token_counter.router, tags=["Token Counter"])
 api_v1_router.include_router(token_generator.router, tags=["Token Generator"])
 api_v1_router.include_router(chat.router, tags=["Chat"])
+api_v1_router.include_router(vector_stores.router, tags=["Vector Stores"])

app/api/v1/vector_stores.py

@@ -0,0 +1,355 @@
+from __future__ import annotations
+
+import asyncio
+import os
+import tempfile
+import time
+
+from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
+
+from app.api.deps import get_vector_store_service, require_auth
+from app.core.logger import get_logger
+from app.models.domain import ConversionError
+from app.models.schemas import (
+    DeleteRequest,
+    DeleteResponse,
+    DocumentIngestRequest,
+    DocumentIngestResponse,
+    SearchRequest,
+    SearchResponse,
+    VectorStoreCreate,
+    VectorStoreListResponse,
+    VectorStoreResponse,
+)
+from app.services.converter_service import ConverterService
+from app.services.vector_store_service import VectorStoreService
+
+router = APIRouter()
+logger = get_logger(__name__)
+
+
+@router.post(
+    "/vector-stores",
+    response_model=VectorStoreResponse,
+    summary="Create a new vector store",
+    status_code=status.HTTP_201_CREATED,
+)
+async def create_vector_store(
+    body: VectorStoreCreate,
+    token: str = Depends(require_auth),
+    vector_store_service: VectorStoreService = Depends(get_vector_store_service),
+) -> VectorStoreResponse:
+    store_id, app_id = await vector_store_service.create_store(
+        name=body.name,
+        description=body.description or "",
+        metadata=body.metadata,
+    )
+    stats = await vector_store_service.get_store_stats(store_id)
+    return VectorStoreResponse(
+        success=True,
+        vector_store_id=store_id,
+        app_id=app_id,
+        name=body.name,
+        description=body.description,
+        embedding_dimension=stats["embedding_dimension"],
+        document_count=stats["document_count"],
+        created_at=stats["created_at"],
+        metadata=stats["metadata"],
+    )
+
+
+@router.get(
+    "/vector-stores",
+    response_model=VectorStoreListResponse,
+    summary="List all vector stores",
+)
+async def list_vector_stores(
+    token: str = Depends(require_auth),
+    vector_store_service: VectorStoreService = Depends(get_vector_store_service),
+) -> VectorStoreListResponse:
+    records = vector_store_service.list_stores()
+    stores = []
+    for r in records:
+        try:
+            stats = await vector_store_service.get_store_stats(r.store_id)
+        except Exception:
+            stats = {}
+        stores.append(VectorStoreResponse(
+            success=True,
+            vector_store_id=r.store_id,
+            app_id=stats.get("app_id", r.store_id),
+            name=r.name,
+            description=r.description,
+            embedding_dimension=stats.get("embedding_dimension", 0),
+            document_count=stats.get("document_count", 0),
+            created_at=r.created_at,
+            metadata=r.metadata,
+        ))
+    return VectorStoreListResponse(success=True, total=len(stores), stores=stores)
+
+
+@router.get(
+    "/vector-stores/{store_id}",
+    response_model=VectorStoreResponse,
+    summary="Get vector store details",
+)
+async def get_vector_store(
+    store_id: str,
+    token: str = Depends(require_auth),
+    vector_store_service: VectorStoreService = Depends(get_vector_store_service),
+) -> VectorStoreResponse:
+    try:
+        stats = await vector_store_service.get_store_stats(store_id)
+    except ValueError as exc:
+        raise HTTPException(status_code=404, detail=str(exc))
+    return VectorStoreResponse(
+        success=True,
+        vector_store_id=stats["store_id"],
+        app_id=stats["app_id"],
+        name=stats["name"],
+        description=stats["description"],
+        embedding_dimension=stats["embedding_dimension"],
+        document_count=stats["document_count"],
+        created_at=stats["created_at"],
+        metadata=stats["metadata"],
+    )
+
+
+@router.delete(
+    "/vector-stores/{store_id}",
+    response_model=VectorStoreResponse,
+    summary="Delete a vector store and all its data",
+)
+async def delete_vector_store(
+    store_id: str,
+    token: str = Depends(require_auth),
+    vector_store_service: VectorStoreService = Depends(get_vector_store_service),
+) -> VectorStoreResponse:
+    record = vector_store_service.get_store(store_id)
+    if record is None:
+        raise HTTPException(status_code=404, detail=f"Vector store {store_id} not found")
+    ok = await vector_store_service.delete_store(store_id)
+    if not ok:
+        raise HTTPException(status_code=500, detail="Failed to delete vector store")
+    return VectorStoreResponse(
+        success=True,
+        vector_store_id=store_id,
+        app_id="",
+        name=record.name,
+        document_count=0,
+        embedding_dimension=0,
+        created_at=record.created_at,
+    )
+
+
+@router.post(
+    "/vector-stores/{store_id}/documents",
+    response_model=DocumentIngestResponse,
+    summary="Ingest a document into the vector store with automatic chunking and embedding",
+)
+async def ingest_document(
+    store_id: str,
+    body: DocumentIngestRequest,
+    token: str = Depends(require_auth),
+    vector_store_service: VectorStoreService = Depends(get_vector_store_service),
+) -> DocumentIngestResponse:
+    record = vector_store_service.get_store(store_id)
+    if record is None:
+        raise HTTPException(status_code=404, detail=f"Vector store {store_id} not found")
+    try:
+        chunks, elapsed = await vector_store_service.ingest_document(
+            store_id=store_id,
+            doc_id=body.doc_id,
+            text=body.text,
+            source=body.source,
+            metadata=body.metadata,
+            chunk_size=body.chunk_size,
+            chunk_overlap=body.chunk_overlap,
+        )
+    except Exception as exc:
+        logger.error("Ingest failed for store %s: %s", store_id, exc)
+        return DocumentIngestResponse(
+            success=False,
+            vector_store_id=store_id,
+            doc_id=body.doc_id,
+            chunks_ingested=0,
+            time_ms=0,
+            error=str(exc),
+        )
+    return DocumentIngestResponse(
+        success=True,
+        vector_store_id=store_id,
+        doc_id=body.doc_id,
+        chunks_ingested=chunks,
+        time_ms=round(elapsed, 3),
+    )
+
+
+@router.post(
+    "/vector-stores/{store_id}/documents/upload",
+    response_model=DocumentIngestResponse,
+    summary="Upload a PDF file and ingest it into the vector store",
+)
+async def ingest_pdf_document(
+    store_id: str,
+    file: UploadFile = File(..., description="PDF file to ingest"),
+    doc_id: str = Form(..., min_length=1, max_length=256),
+    chunk_size: int = Form(512, ge=64, le=4096),
+    chunk_overlap: int = Form(64, ge=0, le=512),
+    token: str = Depends(require_auth),
+    vector_store_service: VectorStoreService = Depends(get_vector_store_service),
+) -> DocumentIngestResponse:
+    record = vector_store_service.get_store(store_id)
+    if record is None:
+        raise HTTPException(status_code=404, detail=f"Vector store {store_id} not found")
+
+    if file.content_type != "application/pdf":
+        raise HTTPException(status_code=400, detail="Only PDF files are accepted")
+    if not file.filename or not file.filename.lower().endswith(".pdf"):
+        raise HTTPException(status_code=400, detail="Only .pdf files are accepted")
+
+    tmp_path = None
+    try:
+        raw = await file.read()
+        if len(raw) < 5 or raw[:5] != b"%PDF-":
+            raise HTTPException(status_code=400, detail="File is not a valid PDF")
+
+        suffix = ".pdf"
+        with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
+            tmp.write(raw)
+            tmp_path = tmp.name
+
+        converter = ConverterService()
+        loop = asyncio.get_running_loop()
+        result = await loop.run_in_executor(None, converter.convert_file, tmp_path)
+
+        if isinstance(result, ConversionError):
+            logger.error("PDF conversion failed: %s", result.message)
+            return DocumentIngestResponse(
+                success=False,
+                vector_store_id=store_id,
+                doc_id=doc_id,
+                chunks_ingested=0,
+                time_ms=0,
+                error=result.message,
+            )
+
+        text = result.markdown
+        source = file.filename
+
+        chunks, elapsed = await vector_store_service.ingest_document(
+            store_id=store_id,
+            doc_id=doc_id,
+            text=text,
+            source=source,
+            chunk_size=chunk_size,
+            chunk_overlap=chunk_overlap,
+        )
+
+        return DocumentIngestResponse(
+            success=True,
+            vector_store_id=store_id,
+            doc_id=doc_id,
+            chunks_ingested=chunks,
+            time_ms=round(elapsed, 3),
+        )
+    except HTTPException:
+        raise
+    except Exception as exc:
+        logger.error("PDF ingest failed for store %s: %s", store_id, exc)
+        return DocumentIngestResponse(
+            success=False,
+            vector_store_id=store_id,
+            doc_id=doc_id,
+            chunks_ingested=0,
+            time_ms=0,
+            error=str(exc),
+        )
+    finally:
+        if tmp_path and os.path.exists(tmp_path):
+            os.unlink(tmp_path)
+        await file.close()
+
+
+@router.post(
+    "/vector-stores/{store_id}/search",
+    response_model=SearchResponse,
+    summary="Search the vector store using natural language queries (RAG)",
+)
+async def search_vector_store(
+    store_id: str,
+    body: SearchRequest,
+    token: str = Depends(require_auth),
+    vector_store_service: VectorStoreService = Depends(get_vector_store_service),
+) -> SearchResponse:
+    record = vector_store_service.get_store(store_id)
+    if record is None:
+        raise HTTPException(status_code=404, detail=f"Vector store {store_id} not found")
+    try:
+        items, elapsed = await vector_store_service.search(
+            store_id=store_id,
+            query_text=body.query,
+            top_k=body.top_k,
+            filter_expr=body.filter,
+            min_score=body.min_score,
+            include_vectors=body.include_vectors,
+            include_metadata=body.include_metadata,
+        )
+    except Exception as exc:
+        logger.error("Search failed for store %s: %s", store_id, exc)
+        return SearchResponse(
+            success=False,
+            vector_store_id=store_id,
+            query=body.query,
+            results=[],
+            total_results=0,
+            time_ms=0,
+            error=str(exc),
+        )
+    return SearchResponse(
+        success=True,
+        vector_store_id=store_id,
+        query=body.query,
+        results=items,
+        total_results=len(items),
+        time_ms=round(elapsed, 3),
+    )
+
+
+@router.post(
+    "/vector-stores/{store_id}/delete",
+    response_model=DeleteResponse,
+    summary="Delete documents from the vector store by IDs or filter",
+)
+async def delete_documents(
+    store_id: str,
+    body: DeleteRequest,
+    token: str = Depends(require_auth),
+    vector_store_service: VectorStoreService = Depends(get_vector_store_service),
+) -> DeleteResponse:
+    record = vector_store_service.get_store(store_id)
+    if record is None:
+        raise HTTPException(status_code=404, detail=f"Vector store {store_id} not found")
+    start = time.perf_counter()
+    try:
+        deleted = await vector_store_service.delete_documents(
+            store_id=store_id,
+            ids=body.ids,
+            filter_expr=body.filter,
+        )
+    except Exception as exc:
+        elapsed = (time.perf_counter() - start) * 1000
+        return DeleteResponse(
+            success=False,
+            vector_store_id=store_id,
+            deleted_count=0,
+            time_ms=round(elapsed, 3),
+            error=str(exc),
+        )
+    elapsed = (time.perf_counter() - start) * 1000
+    return DeleteResponse(
+        success=True,
+        vector_store_id=store_id,
+        deleted_count=deleted,
+        time_ms=round(elapsed, 3),
+    )

app/config.py

@@ -61,6 +61,13 @@ class Settings(BaseSettings):
     key_lock_ttl: int = 600
     rounds_per_model: int = 50
 
+    data_dir: str = "./data"
+    embedding_model: str = "ibm-granite/granite-embedding-small-english-r2"
+    embedding_dimension: int = 384
+    default_top_k: int = 10
+    max_chunk_size: int = 512
+    chunk_overlap: int = 64
+
     jwt_secret_key: str = "changeme-jwt-secret"
     jwt_algorithm: str = "HS256"
     jwt_default_expiry_minutes: int = 30

app/core/vector_store/__init__.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+from app.core.vector_store.models import VectorStoreIndex, Base
+from app.core.vector_store.deps import (
+    engine,
+    AsyncSessionLocal,
+    init_vector_store_db,
+    get_vs_db,
+)
+
+__all__ = [
+    "Base",
+    "VectorStoreIndex",
+    "engine",
+    "AsyncSessionLocal",
+    "init_vector_store_db",
+    "get_vs_db",
+]

app/core/vector_store/deps.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+import os
+from typing import AsyncGenerator
+
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+from app.config import get_settings
+from app.core.vector_store.models import Base
+
+_settings = get_settings()
+
+DATA_DIR = _settings.data_dir
+os.makedirs(DATA_DIR, exist_ok=True)
+
+VECTOR_STORE_DB_URL = f"sqlite+aiosqlite:///{os.path.join(DATA_DIR, 'vector_stores.db')}"
+
+engine = create_async_engine(VECTOR_STORE_DB_URL, echo=False)
+AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+
+
+async def init_vector_store_db():
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+
+async def get_vs_db() -> AsyncGenerator[AsyncSession, None]:
+    async with AsyncSessionLocal() as session:
+        try:
+            yield session
+        finally:
+            await session.close()

app/core/vector_store/models.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from typing import Any, Dict, Optional
+
+from sqlalchemy import Column, String, Text
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+def _utcnow() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+class VectorStoreIndex(Base):
+    __tablename__ = "vector_store_index"
+
+    store_id: Mapped[str] = mapped_column(String(36), primary_key=True)
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
+    path: Mapped[str] = mapped_column(String(1024), nullable=False)
+    description: Mapped[str] = mapped_column(String(1024), default="")
+    metadata_json: Mapped[str] = mapped_column(Text, default="{}")
+    created_at: Mapped[str] = mapped_column(String(64), default=_utcnow)
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "store_id": self.store_id,
+            "name": self.name,
+            "path": self.path,
+            "description": self.description,
+            "metadata": json.loads(self.metadata_json or "{}"),
+            "created_at": self.created_at,
+        }
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "VectorStoreIndex":
+        return cls(
+            store_id=data["store_id"],
+            name=data["name"],
+            path=data["path"],
+            description=data.get("description", ""),
+            metadata_json=json.dumps(data.get("metadata", {})),
+            created_at=data.get("created_at", _utcnow()),
+        )

app/models/schemas.py

@@ -563,3 +563,92 @@ class SqlValidationResponse(BaseModel):
     warnings: List[str] = []
     tables: List[str] = []
     columns: List[str] = []
+
+
+# ---------------------------------------------------------------------------
+# Vector Store (RAG) - Powered by Zvec
+# ---------------------------------------------------------------------------
+
+class VectorStoreCreate(BaseModel):
+    name: str = Field(..., min_length=1, max_length=256, description="Human-readable name")
+    description: Optional[str] = Field(None, max_length=1024)
+    metadata: Dict[str, Any] = Field(default_factory=dict)
+
+
+class VectorStoreResponse(BaseModel):
+    success: bool
+    vector_store_id: str
+    app_id: str
+    name: str
+    description: Optional[str] = None
+    embedding_dimension: int
+    document_count: int
+    created_at: str
+    metadata: Dict[str, Any] = {}
+
+
+class VectorStoreListResponse(BaseModel):
+    success: bool
+    total: int
+    stores: List[VectorStoreResponse]
+
+
+class DocumentIngestRequest(BaseModel):
+    doc_id: str = Field(..., min_length=1, max_length=256)
+    text: str = Field(..., min_length=1)
+    source: Optional[str] = Field(None, max_length=512)
+    metadata: Dict[str, Any] = Field(default_factory=dict)
+    chunk_size: Optional[int] = Field(None, ge=64, le=4096)
+    chunk_overlap: Optional[int] = Field(None, ge=0, le=512)
+
+
+class DocumentIngestResponse(BaseModel):
+    success: bool
+    vector_store_id: str
+    doc_id: str
+    chunks_ingested: int
+    time_ms: float
+    error: Optional[str] = None
+
+
+class SearchRequest(BaseModel):
+    query: str = Field(..., min_length=1, max_length=5000, description="Natural language query")
+    top_k: int = Field(default=10, ge=1, le=100, description="Max results to return")
+    filter: Optional[str] = Field(None, description="Filter expression (e.g. 'category = \"tech\"')")
+    min_score: Optional[float] = Field(None, ge=0.0, le=1.0, description="Minimum similarity score threshold")
+    include_vectors: bool = Field(default=False, description="Include vector embeddings in results")
+    include_metadata: bool = Field(default=False, description="Include source metadata in results")
+
+
+class SearchResultItem(BaseModel):
+    rank: int
+    doc_id: str
+    chunk_index: int
+    text: str
+    score: float
+    source: Optional[str] = None
+    metadata: Dict[str, Any] = {}
+    vector: Optional[List[float]] = None
+
+
+class SearchResponse(BaseModel):
+    success: bool
+    vector_store_id: str
+    query: str
+    results: List[SearchResultItem]
+    total_results: int
+    time_ms: float
+    error: Optional[str] = None
+
+
+class DeleteRequest(BaseModel):
+    ids: Optional[List[str]] = Field(None)
+    filter: Optional[str] = Field(None)
+
+
+class DeleteResponse(BaseModel):
+    success: bool
+    vector_store_id: str
+    deleted_count: int
+    time_ms: float
+    error: Optional[str] = None

app/services/chunking_service.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import re
+from typing import List
+
+logger = logging.getLogger(__name__)
+
+
+def chunk_text(
+    text: str,
+    chunk_size: int = 512,
+    chunk_overlap: int = 64,
+) -> List[str]:
+    if chunk_overlap >= chunk_size:
+        chunk_overlap = chunk_size // 4
+
+    paragraphs = re.split(r"\n\s*\n", text.strip())
+    chunks: List[str] = []
+    current: List[str] = []
+    current_len = 0
+
+    for para in paragraphs:
+        para = para.strip()
+        if not para:
+            continue
+        para_len = len(para)
+
+        if current_len + para_len + 1 <= chunk_size:
+            current.append(para)
+            current_len += para_len + 1
+        else:
+            if current:
+                chunks.append("\n\n".join(current))
+            if para_len > chunk_size:
+                for i in range(0, para_len, chunk_size):
+                    segment = para[i:i + chunk_size]
+                    if len(segment) >= chunk_size // 3:
+                        chunks.append(segment)
+                current = []
+                current_len = 0
+            else:
+                current = [para]
+                current_len = para_len + 1
+
+    if current:
+        chunks.append("\n\n".join(current))
+
+    if chunk_overlap > 0 and len(chunks) > 1:
+        overlapped: List[str] = []
+        for i, chunk in enumerate(chunks):
+            if i == 0:
+                overlapped.append(chunk)
+            else:
+                prev = chunks[i - 1]
+                overlap_text = " ".join(prev.split()[-chunk_overlap:]) if len(prev.split()) > chunk_overlap else prev
+                combined = f"{overlap_text}\n\n{chunk}"
+                overlapped.append(combined)
+        return overlapped if all(len(c) <= chunk_size + chunk_overlap + 10 for c in overlapped) else chunks
+
+    return chunks if chunks else [text]
+
+
+async def chunk_text_async(
+    text: str,
+    chunk_size: int = 512,
+    chunk_overlap: int = 64,
+) -> List[str]:
+    loop = asyncio.get_running_loop()
+    return await loop.run_in_executor(None, chunk_text, text, chunk_size, chunk_overlap)

app/services/vector_store_service.py

@@ -0,0 +1,528 @@
+from __future__ import annotations
+
+import asyncio
+import concurrent.futures
+import json
+import os
+import shutil
+import time
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional, Tuple
+
+import zvec
+from sqlalchemy import select
+
+from app.config import get_settings
+from app.core.logger import get_logger
+from app.core.vector_store.deps import AsyncSessionLocal
+from app.core.vector_store.models import VectorStoreIndex
+from app.services.chunking_service import chunk_text_async
+from app.services.embeddings_service import EmbeddingService
+
+logger = get_logger(__name__)
+settings = get_settings()
+
+_EMBEDDING_DIM = 384
+_MAX_WORKERS = min(16, (os.cpu_count() or 1) + 4)
+
+
+def _run_sync(fn, *args, **kwargs):
+    return fn(*args, **kwargs)
+
+
+class VectorStoreRecord:
+    def __init__(
+        self,
+        store_id: str,
+        name: str,
+        path: str,
+        description: str = "",
+        metadata: Optional[Dict[str, Any]] = None,
+        created_at: Optional[str] = None,
+    ):
+        self.store_id = store_id
+        self.name = name
+        self.path = path
+        self.description = description
+        self.metadata = metadata or {}
+        self.created_at = created_at or datetime.now(timezone.utc).isoformat()
+
+
+class VectorStoreService:
+    def __init__(self, embedding_service: EmbeddingService):
+        self._embedding_service = embedding_service
+        self._stores: Dict[str, VectorStoreRecord] = {}
+        self._collections: Dict[str, "zvec.Collection"] = {}
+        self._data_dir = os.path.join(settings.data_dir, "vector_stores")
+        self._thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=_MAX_WORKERS, thread_name_prefix="zvec")
+        os.makedirs(self._data_dir, exist_ok=True)
+
+    def _store_path(self, store_id: str) -> str:
+        return os.path.join(self._data_dir, f"store_{store_id}")
+
+    def _get_collection(self, store_id: str) -> Optional["zvec.Collection"]:
+        return self._collections.get(store_id)
+
+    # --- SQLite persistence ---
+
+    async def init_db(self) -> None:
+        from app.core.vector_store.deps import init_vector_store_db
+        await init_vector_store_db()
+        async with AsyncSessionLocal() as session:
+            result = await session.execute(select(VectorStoreIndex))
+            rows = result.scalars().all()
+            for row in rows:
+                d = row.to_dict()
+                record = VectorStoreRecord(
+                    store_id=d["store_id"],
+                    name=d["name"],
+                    path=d["path"],
+                    description=d["description"],
+                    metadata=d["metadata"],
+                    created_at=d["created_at"],
+                )
+                self._stores[record.store_id] = record
+                store_path = d["path"]
+                if os.path.exists(os.path.join(store_path, "__zvec_meta")):
+                    try:
+                        col = zvec.open(store_path)
+                        if col is not None:
+                            self._collections[record.store_id] = col
+                    except Exception as exc:
+                        logger.warning("Could not open collection %s: %s", record.store_id, exc)
+
+    async def _persist_store(self, record: VectorStoreRecord) -> None:
+        async with AsyncSessionLocal() as session:
+            existing = await session.get(VectorStoreIndex, record.store_id)
+            if existing:
+                existing.name = record.name
+                existing.description = record.description
+                existing.metadata_json = json.dumps(record.metadata)
+            else:
+                session.add(VectorStoreIndex.from_dict({
+                    "store_id": record.store_id,
+                    "name": record.name,
+                    "path": record.path,
+                    "description": record.description,
+                    "metadata": record.metadata,
+                    "created_at": record.created_at,
+                }))
+            await session.commit()
+
+    async def _remove_persisted_store(self, store_id: str) -> None:
+        async with AsyncSessionLocal() as session:
+            row = await session.get(VectorStoreIndex, store_id)
+            if row:
+                await session.delete(row)
+                await session.commit()
+
+    # --- Synchronous helpers (run in thread pool) ---
+
+    def _open_or_create_collection_sync(self, store_id: str, store_path: str) -> "zvec.Collection":
+        col = self._collections.get(store_id)
+        if col is not None:
+            return col
+
+        if os.path.exists(os.path.join(store_path, "__zvec_meta")):
+            logger.info("Opening existing collection: %s", store_id)
+            try:
+                col = zvec.open(store_path)
+                if col is not None:
+                    self._collections[store_id] = col
+                    return col
+            except Exception as exc:
+                logger.warning("Could not open collection %s: %s", store_id, exc)
+
+        schema = zvec.CollectionSchema(
+            name=f"store_{store_id}",
+            fields=[
+                zvec.FieldSchema(name="text", data_type=zvec.DataType.STRING),
+                zvec.FieldSchema(
+                    name="doc_id",
+                    data_type=zvec.DataType.STRING,
+                    index_param=zvec.InvertIndexParam(),
+                ),
+                zvec.FieldSchema(name="chunk_index", data_type=zvec.DataType.INT32),
+                zvec.FieldSchema(
+                    name="source",
+                    data_type=zvec.DataType.STRING,
+                    index_param=zvec.InvertIndexParam(),
+                ),
+                zvec.FieldSchema(name="created_at", data_type=zvec.DataType.INT64),
+            ],
+            vectors=[
+                zvec.VectorSchema(
+                    name="embedding",
+                    data_type=zvec.DataType.VECTOR_FP32,
+                    dimension=_EMBEDDING_DIM,
+                    index_param=zvec.HnswIndexParam(metric_type=zvec.MetricType.COSINE),
+                ),
+            ],
+        )
+        logger.info("Creating new collection: %s", store_id)
+        col = zvec.create_and_open(path=store_path, schema=schema)
+        self._collections[store_id] = col
+        return col
+
+    def _ingest_document_sync(
+        self,
+        store_id: str,
+        doc_id: str,
+        text: str,
+        chunks: List[str],
+        embeddings: List[List[float]],
+        source: str,
+    ) -> Tuple[int, float]:
+        col = self._get_collection(store_id)
+        if col is None:
+            record = self._stores.get(store_id)
+            if record is None:
+                raise ValueError(f"Vector store {store_id} not found")
+            col = self._open_or_create_collection_sync(store_id, record.path)
+
+        now_ts = int(time.time())
+
+        docs = [
+            zvec.Doc(
+                id=f"{doc_id}_{i}",
+                vectors={"embedding": emb},
+                fields={
+                    "text": chunk_text_str,
+                    "doc_id": doc_id,
+                    "chunk_index": i,
+                    "source": source or "",
+                    "created_at": now_ts,
+                },
+            )
+            for i, (chunk_text_str, emb) in enumerate(zip(chunks, embeddings))
+        ]
+
+        for i in range(0, len(docs), 100):
+            col.insert(docs[i:i + 100])
+
+        col.flush()
+        col.optimize()
+        return len(chunks), 0.0
+
+    def _search_sync(
+        self,
+        store_id: str,
+        query_emb: List[float],
+        top_k: int,
+        filter_expr: Optional[str],
+        min_score: Optional[float] = None,
+        include_vectors: bool = False,
+        include_metadata: bool = False,
+    ) -> List[Dict[str, Any]]:
+        col = self._get_collection(store_id)
+        if col is None:
+            record = self._stores.get(store_id)
+            if record is None:
+                raise ValueError(f"Vector store {store_id} not found")
+            col = self._open_or_create_collection_sync(store_id, record.path)
+
+        kwargs: Dict[str, Any] = {
+            "vectors": zvec.VectorQuery(field_name="embedding", vector=query_emb),
+            "topk": top_k,
+        }
+        if filter_expr:
+            kwargs["filter"] = filter_expr
+
+        results = col.query(**kwargs)
+
+        items = []
+        for i, r in enumerate(results):
+            score = float(r.score) if hasattr(r, "score") and r.score is not None else 0.0
+            if min_score is not None and score > (1.0 - min_score):
+                continue
+            item: Dict[str, Any] = {
+                "rank": len(items) + 1,
+                "doc_id": r.field("doc_id") if hasattr(r, "field") else r.id.rsplit("_", 1)[0],
+                "chunk_index": r.field("chunk_index") if hasattr(r, "field") else 0,
+                "text": r.field("text") if hasattr(r, "field") else "",
+                "score": score,
+                "source": r.field("source") if hasattr(r, "field") else "",
+                "metadata": {},
+                "vector": None,
+            }
+            if include_vectors:
+                try:
+                    vec = r.vector("embedding") if hasattr(r, "vector") else None
+                    item["vector"] = list(vec) if vec is not None else None
+                except Exception:
+                    item["vector"] = None
+            if include_metadata:
+                item["metadata"] = {
+                    "doc_id": item["doc_id"],
+                    "chunk_index": item["chunk_index"],
+                    "source": item["source"],
+                }
+            items.append(item)
+        return items
+
+    def _fetch_documents_sync(self, store_id: str, ids: List[str]) -> Dict[str, Any]:
+        col = self._get_collection(store_id)
+        if col is None:
+            record = self._stores.get(store_id)
+            if record is None:
+                raise ValueError(f"Vector store {store_id} not found")
+            col = self._open_or_create_collection_sync(store_id, record.path)
+
+        internal_ids = []
+        for doc_id in ids:
+            fetched = col.fetch(ids=[doc_id])
+            if doc_id in fetched:
+                internal_ids.append(doc_id)
+                continue
+            for i in range(0, 1024):
+                chunk_id = f"{doc_id}_{i}"
+                fetched = col.fetch(ids=[chunk_id])
+                if chunk_id in fetched:
+                    internal_ids.append(chunk_id)
+                else:
+                    break
+            break
+
+        if not internal_ids:
+            return {}
+
+        fetched = col.fetch(ids=internal_ids)
+        result = {}
+        for k, v in fetched.items():
+            result[k] = {
+                "id": v.id,
+                "text": v.field("text") if hasattr(v, "field") else "",
+                "doc_id": v.field("doc_id") if hasattr(v, "field") else "",
+                "chunk_index": v.field("chunk_index") if hasattr(v, "field") else 0,
+                "source": v.field("source") if hasattr(v, "field") else "",
+            }
+        return result
+
+    def _delete_documents_sync(
+        self,
+        store_id: str,
+        ids: Optional[List[str]],
+        filter_expr: Optional[str],
+    ) -> int:
+        col = self._get_collection(store_id)
+        if col is None:
+            record = self._stores.get(store_id)
+            if record is None:
+                raise ValueError(f"Vector store {store_id} not found")
+            col = self._open_or_create_collection_sync(store_id, record.path)
+
+        deleted = 0
+        if ids:
+            chunk_ids = []
+            for doc_id in ids:
+                chunk_ids.extend(f"{doc_id}_{i}" for i in range(4096))
+            fetched = col.fetch(ids=chunk_ids[:1000])
+            actual_ids = [k for k in chunk_ids if k in fetched]
+            if actual_ids:
+                col.delete(ids=actual_ids)
+                deleted = len(actual_ids)
+
+        if filter_expr:
+            col.delete_by_filter(filter=filter_expr)
+            deleted = max(deleted, 1)
+
+        col.flush()
+        return deleted
+
+    def _get_store_stats_sync(self, store_id: str) -> Dict[str, Any]:
+        col = self._get_collection(store_id)
+        if col is None:
+            record = self._stores.get(store_id)
+            if record is None:
+                raise ValueError(f"Vector store {store_id} not found")
+            col = self._open_or_create_collection_sync(store_id, record.path)
+
+        try:
+            stats = col.stats
+            doc_count = getattr(stats, 'doc_count', 0) if stats else 0
+        except Exception:
+            doc_count = 0
+
+        record = self._stores[store_id]
+        return {
+            "store_id": store_id,
+            "name": record.name,
+            "description": record.description,
+            "app_id": settings.application_id or store_id,
+            "embedding_dimension": _EMBEDDING_DIM,
+            "document_count": doc_count,
+            "created_at": record.created_at,
+            "metadata": record.metadata,
+        }
+
+    # --- Async public API ---
+
+    async def _run_in_thread(self, fn, *args, **kwargs):
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(self._thread_pool, _run_sync, lambda: fn(*args, **kwargs))
+
+    async def _run_sync_fn(self, fn):
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(self._thread_pool, fn)
+
+    async def create_store(
+        self,
+        name: str,
+        description: str = "",
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> Tuple[str, str]:
+        store_id = str(uuid.uuid4())
+        store_path = self._store_path(store_id)
+
+        await self._run_sync_fn(
+            lambda: self._open_or_create_collection_sync(store_id, store_path)
+        )
+
+        record = VectorStoreRecord(
+            store_id=store_id,
+            name=name,
+            path=store_path,
+            description=description,
+            metadata=metadata or {},
+        )
+        self._stores[store_id] = record
+        await self._persist_store(record)
+
+        app_id = settings.application_id or store_id
+        logger.info("Created vector store: %s (name=%s, app_id=%s)", store_id, name, app_id)
+        return store_id, app_id
+
+    def list_stores(self) -> List[VectorStoreRecord]:
+        return list(self._stores.values())
+
+    def get_store(self, store_id: str) -> Optional[VectorStoreRecord]:
+        return self._stores.get(store_id)
+
+    async def delete_store(self, store_id: str) -> bool:
+        record = self._stores.pop(store_id, None)
+        if record is None:
+            return False
+
+        col = self._collections.pop(store_id, None)
+        if col is not None:
+            try:
+                await self._run_sync_fn(lambda: col.destroy())
+            except Exception as exc:
+                logger.warning("Error destroying collection %s: %s", store_id, exc)
+
+        store_path = record.path
+        if os.path.exists(store_path):
+            await self._run_sync_fn(lambda: shutil.rmtree(store_path, ignore_errors=True))
+
+        await self._remove_persisted_store(store_id)
+        logger.info("Deleted vector store: %s", store_id)
+        return True
+
+    async def _ensure_embedding_model(self) -> None:
+        if not self._embedding_service.is_loaded(_EMBEDDING_DIM):
+            loop = asyncio.get_running_loop()
+            await loop.run_in_executor(None, self._embedding_service.load_model, _EMBEDDING_DIM)
+
+    async def ingest_document(
+        self,
+        store_id: str,
+        doc_id: str,
+        text: str,
+        source: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+        chunk_size: Optional[int] = None,
+        chunk_overlap: Optional[int] = None,
+    ) -> Tuple[int, float]:
+        await self._ensure_embedding_model()
+
+        size = chunk_size or 512
+        overlap = chunk_overlap or 64
+        chunks = await chunk_text_async(text, chunk_size=size, chunk_overlap=overlap)
+        metadata = metadata or {}
+        source = source or metadata.get("source", "")
+
+        start = time.perf_counter()
+
+        loop = asyncio.get_running_loop()
+        embeddings = await loop.run_in_executor(
+            self._thread_pool,
+            self._embedding_service.generate_embedding,
+            chunks,
+            _EMBEDDING_DIM,
+        )
+
+        elapsed_sync = await self._run_sync_fn(
+            lambda: self._ingest_document_sync(store_id, doc_id, text, chunks, embeddings, source or "")
+        )
+
+        elapsed = (time.perf_counter() - start) * 1000
+        logger.info("Ingested doc %s into store %s: %d chunks in %.2f ms", doc_id, store_id, len(chunks), elapsed)
+        return len(chunks), elapsed
+
+    async def search(
+        self,
+        store_id: str,
+        query_text: str,
+        top_k: int = 10,
+        filter_expr: Optional[str] = None,
+        min_score: Optional[float] = None,
+        include_vectors: bool = False,
+        include_metadata: bool = False,
+    ) -> Tuple[List[Dict[str, Any]], float]:
+        await self._ensure_embedding_model()
+
+        start = time.perf_counter()
+
+        loop = asyncio.get_running_loop()
+        query_emb = await loop.run_in_executor(
+            self._thread_pool,
+            lambda: self._embedding_service.generate_embedding([query_text], _EMBEDDING_DIM)[0],
+        )
+
+        items = await self._run_sync_fn(
+            lambda: self._search_sync(store_id, query_emb, top_k, filter_expr, min_score, include_vectors, include_metadata)
+        )
+
+        elapsed = (time.perf_counter() - start) * 1000
+        return items, elapsed
+
+    async def fetch_documents(self, store_id: str, ids: List[str]) -> Dict[str, Any]:
+        return await self._run_sync_fn(
+            lambda: self._fetch_documents_sync(store_id, ids)
+        )
+
+    async def delete_documents(
+        self,
+        store_id: str,
+        ids: Optional[List[str]] = None,
+        filter_expr: Optional[str] = None,
+    ) -> int:
+        return await self._run_sync_fn(
+            lambda: self._delete_documents_sync(store_id, ids, filter_expr)
+        )
+
+    async def get_store_stats(self, store_id: str) -> Dict[str, Any]:
+        return await self._run_sync_fn(
+            lambda: self._get_store_stats_sync(store_id)
+        )
+
+    async def get_total_document_count(self) -> int:
+        total = 0
+        for store_id in list(self._stores.keys()):
+            try:
+                stats = await self.get_store_stats(store_id)
+                total += stats.get("document_count", 0)
+            except Exception:
+                pass
+        return total
+
+    async def close_all(self) -> None:
+        for store_id, col in list(self._collections.items()):
+            try:
+                await self._run_sync_fn(lambda: col.flush())
+            except Exception:
+                pass
+        self._collections.clear()
+        self._stores.clear()
+        self._thread_pool.shutdown(wait=True)

requirements.txt

@@ -1,3 +1,4 @@
+zvec>=0.4.0
 markitdown[all]>=0.1.5
 fastapi>=0.111.0
 uvicorn[standard]>=0.30.0