feat: Implement RAG document and sync services with vector store integration

- Added `vectorstore.py` for managing vector storage operations using Milvus.
- Introduced `rag_document.py` for handling RAG document lifecycle, including ingestion and status updates.
- Created `rag_status.py` for streaming RAG ingestion status via Redis pub/sub.
- Developed `rag_sync.py` for managing synchronization operations and logs.
- Implemented `sync_source.py` for managing sync source configurations and triggering syncs.
- Established a lightweight task dispatcher in `dispatcher.py` for background task execution.
- Registered task functions for RAG ingestion and sync operations in `rag_tasks.py`.
- Configured task dispatcher settings in `arq_settings.py`.
- Added necessary database models and schemas for RAG document and sync operations.
This commit is contained in:
2026-06-13 21:52:27 +08:00
parent 75e276efc7
commit bb4b614711
43 changed files with 6151 additions and 9 deletions
+2
View File
@@ -16,6 +16,7 @@ from backend.routes.admin import router as admin_router
from backend.routes.auth import router as auth_router
from backend.routes.chat import router as chat_router
from backend.routes.health import router as health_router
from backend.routes.rag import router as rag_router
from backend.routes.users import router as users_router
logger = logging.getLogger("agent_alpha")
@@ -86,6 +87,7 @@ class AppBuilder:
app.include_router(health_router)
app.include_router(chat_router)
app.include_router(users_router)
app.include_router(rag_router)
return app
+55
View File
@@ -2,6 +2,9 @@
from __future__ import annotations
from pathlib import Path
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -30,6 +33,58 @@ class Settings(BaseSettings):
database_url: str = "postgresql+asyncpg://agent_alpha:agent_alpha@localhost:5432/agent_alpha"
valkey_url: str = "redis://localhost:6379/0"
# ── Milvus (Vector Database) ───────────────────────────────────────────
milvus_uri: str = "http://localhost:19530"
milvus_token: str = ""
# ── Redis pub/sub (for RAG status SSE) ─────────────────────────────────
redis_host: str = "localhost"
redis_port: int = 6379
redis_db: int = 0
# ── Media & File Storage ───────────────────────────────────────────────
media_dir: str = "media"
max_upload_size_mb: int = 50
# ── RAG Parsing ────────────────────────────────────────────────────────
pdf_parser: str = "pymupdf"
# ── Cross-Encoder Reranker ─────────────────────────────────────────────
cross_encoder_model: str = "cross-encoder/ms-marco-MiniLM-L6-v2"
hf_token: str = ""
models_cache_dir: Path = Path.home() / ".cache" / "agent-alpha" / "models"
# ── AI Configuration ───────────────────────────────────────────────────
ai_model: str = "gpt-4o"
rag_image_description_model: str = ""
# ── RAG Settings (derived) ────────────────────────────────────────────
@property
def rag(self) -> object:
"""Return a ``RAGSettings`` instance configured from application settings.
Lazy import to avoid circular dependencies at module level.
"""
from backend.services.rag.config import RAGSettings
return RAGSettings(
collection_name="documents",
chunk_size=512,
chunk_overlap=50,
chunking_strategy="recursive",
enable_hybrid_search=False,
enable_ocr=False,
enable_image_description=True,
image_description_model=self.rag_image_description_model,
)
# ── Factories ──────────────────────────────────────────────────────────
@classmethod
+2
View File
@@ -45,6 +45,8 @@ from sqlalchemy.ext.asyncio import (
from backend.core.config import settings
from backend.core.models import Base, User
# Import RAG models so they are registered with Base.metadata for create_all
from backend.db.models import RAGDocument, SyncLog, SyncSource, ChatFile # noqa: F401
# ── PostgreSQL (SQLAlchemy async) ──────────────────────────────────────────
+36
View File
@@ -0,0 +1,36 @@
"""Custom exceptions for Agent Alpha."""
from __future__ import annotations
class AgentAlphaException(Exception):
"""Base exception for application-specific errors."""
def __init__(
self,
message: str = "An error occurred",
details: dict | None = None,
) -> None:
self.message = message
self.details = details or {}
super().__init__(self.message)
@property
def status_code(self) -> int:
return 500
class NotFoundError(AgentAlphaException):
"""Raised when a requested resource does not exist."""
@property
def status_code(self) -> int:
return 404
class BadRequestError(AgentAlphaException):
"""Raised when the request is invalid."""
@property
def status_code(self) -> int:
return 400
View File
+8
View File
@@ -0,0 +1,8 @@
"""SQLAlchemy ORM models for Agent Alpha."""
from backend.db.models.rag_document import RAGDocument
from backend.db.models.sync_log import SyncLog
from backend.db.models.sync_source import SyncSource
from backend.db.models.chat_file import ChatFile
__all__ = ["RAGDocument", "SyncLog", "SyncSource", "ChatFile"]
+45
View File
@@ -0,0 +1,45 @@
"""Chat file ORM model."""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from sqlalchemy import DateTime, Integer, String, Text
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column
from backend.core.models import Base
class ChatFile(Base):
"""Tracks a file uploaded by a user in a chat session."""
__tablename__ = "chat_files"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
)
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), nullable=False
)
filename: Mapped[str] = mapped_column(String(512), nullable=False)
mime_type: Mapped[str] = mapped_column(String(255), nullable=False, default="")
size: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
storage_path: Mapped[str] = mapped_column(
String(1024), nullable=False, default=""
)
file_type: Mapped[str] = mapped_column(
String(20), nullable=False, default="text"
)
parsed_content: Mapped[str | None] = mapped_column(
Text, nullable=True, default=None
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
default=lambda: datetime.now(timezone.utc),
)
def __repr__(self) -> str:
return f"<ChatFile {self.filename!r} type={self.file_type!r}>"
+50
View File
@@ -0,0 +1,50 @@
"""RAG document ORM model."""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from sqlalchemy import Boolean, DateTime, Integer, String, Text
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column
from backend.core.models import Base
class RAGDocument(Base):
"""Tracks a document ingested into the RAG pipeline."""
__tablename__ = "rag_documents"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
)
collection_name: Mapped[str] = mapped_column(
String(255), nullable=False, default="documents"
)
filename: Mapped[str] = mapped_column(String(512), nullable=False)
filesize: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
filetype: Mapped[str] = mapped_column(String(50), nullable=False, default="")
status: Mapped[str] = mapped_column(
String(20), nullable=False, default="processing"
)
error_message: Mapped[str | None] = mapped_column(Text, nullable=True, default=None)
vector_document_id: Mapped[str | None] = mapped_column(
String(255), nullable=True, default=None
)
chunk_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
storage_path: Mapped[str] = mapped_column(
String(1024), nullable=False, default=""
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
default=lambda: datetime.now(timezone.utc),
)
completed_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True, default=None
)
def __repr__(self) -> str:
return f"<RAGDocument {self.filename!r} status={self.status!r}>"
+50
View File
@@ -0,0 +1,50 @@
"""Sync log ORM model."""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from sqlalchemy import DateTime, Integer, String, Text
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column
from backend.core.models import Base
class SyncLog(Base):
"""Logs a RAG sync operation."""
__tablename__ = "sync_logs"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
)
source: Mapped[str] = mapped_column(String(100), nullable=False, default="")
collection_name: Mapped[str] = mapped_column(
String(255), nullable=False, default="documents"
)
status: Mapped[str] = mapped_column(
String(20), nullable=False, default="running"
)
mode: Mapped[str] = mapped_column(
String(20), nullable=False, default="full"
)
total_files: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
ingested: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
updated: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
skipped: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
failed: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
error_message: Mapped[str | None] = mapped_column(Text, nullable=True, default=None)
sync_source_id: Mapped[uuid.UUID | None] = mapped_column(
UUID(as_uuid=True), nullable=True, default=None
)
started_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True, default=None
)
completed_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True, default=None
)
def __repr__(self) -> str:
return f"<SyncLog {self.source!r} status={self.status!r}>"
+56
View File
@@ -0,0 +1,56 @@
"""Sync source ORM model."""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from sqlalchemy import Boolean, DateTime, Integer, String, Text
from sqlalchemy.dialects.postgresql import JSON, UUID
from sqlalchemy.orm import Mapped, mapped_column
from backend.core.models import Base
class SyncSource(Base):
"""Configuration for a RAG sync source (local folder, GDrive, S3, etc.)."""
__tablename__ = "sync_sources"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
)
name: Mapped[str] = mapped_column(String(255), nullable=False)
connector_type: Mapped[str] = mapped_column(
String(50), nullable=False
)
collection_name: Mapped[str] = mapped_column(
String(255), nullable=False, default="documents"
)
config: Mapped[dict | None] = mapped_column(JSON, nullable=True, default=None)
sync_mode: Mapped[str] = mapped_column(
String(20), nullable=False, default="full"
)
schedule_minutes: Mapped[int | None] = mapped_column(
Integer, nullable=True, default=None
)
is_active: Mapped[bool] = mapped_column(
Boolean, nullable=False, default=True
)
last_sync_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True, default=None
)
last_sync_status: Mapped[str | None] = mapped_column(
String(20), nullable=True, default=None
)
last_error: Mapped[str | None] = mapped_column(
Text, nullable=True, default=None
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
default=lambda: datetime.now(timezone.utc),
)
def __repr__(self) -> str:
return f"<SyncSource {self.name!r} type={self.connector_type!r}>"
+38
View File
@@ -0,0 +1,38 @@
"""Connector registry for RAG sync sources.
Connectors provide a pluggable way to ingest documents from various sources
(local folders, GDrive, S3, OneDrive, etc.). The registry is populated at
import time by each connector module.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any
class BaseConnector(ABC):
"""Abstract base class for sync source connectors."""
CONNECTOR_TYPE: str = ""
DISPLAY_NAME: str = ""
CONFIG_SCHEMA: dict[str, dict[str, Any]] = {}
@abstractmethod
async def validate_config(self, config: dict[str, Any]) -> tuple[bool, str | None]:
"""Validate connector configuration. Returns (is_valid, error_message)."""
...
@abstractmethod
async def list_files(self, config: dict[str, Any]) -> list[dict[str, Any]]:
"""List files available from the source. Returns list of file info dicts."""
...
@abstractmethod
async def download_file(self, config: dict[str, Any], file_path: str) -> bytes:
"""Download a file from the source as bytes."""
...
# Registry populated by connector modules at import time
CONNECTOR_REGISTRY: dict[str, type[BaseConnector]] = {}
+15
View File
@@ -0,0 +1,15 @@
"""Repository module — async CRUD helpers for DB models."""
from backend.repositories import (
rag_document_repo,
sync_log_repo,
sync_source_repo,
chat_file_repo,
)
__all__ = [
"rag_document_repo",
"sync_log_repo",
"sync_source_repo",
"chat_file_repo",
]
+41
View File
@@ -0,0 +1,41 @@
"""Repository for ChatFile CRUD operations."""
from __future__ import annotations
from uuid import UUID
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from backend.db.models.chat_file import ChatFile
async def get_by_id(session: AsyncSession, file_id: UUID) -> ChatFile | None:
"""Get a chat file by ID."""
return await session.get(ChatFile, file_id)
async def create(
session: AsyncSession,
*,
user_id: UUID,
filename: str,
mime_type: str,
size: int,
storage_path: str,
file_type: str,
parsed_content: str | None = None,
) -> ChatFile:
"""Create a new chat file record."""
chat_file = ChatFile(
user_id=user_id,
filename=filename,
mime_type=mime_type,
size=size,
storage_path=storage_path,
file_type=file_type,
parsed_content=parsed_content,
)
session.add(chat_file)
await session.flush()
return chat_file
+86
View File
@@ -0,0 +1,86 @@
"""Repository for RAGDocument CRUD operations."""
from __future__ import annotations
from datetime import datetime
from uuid import UUID
from sqlalchemy import delete, func, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from backend.db.models.rag_document import RAGDocument
async def get_all(
session: AsyncSession,
collection_name: str | None = None,
) -> list[RAGDocument]:
"""List all RAG documents, optionally filtered by collection."""
stmt = select(RAGDocument).order_by(RAGDocument.created_at.desc())
if collection_name:
stmt = stmt.where(RAGDocument.collection_name == collection_name)
result = await session.execute(stmt)
return list(result.scalars().all())
async def get_by_id(session: AsyncSession, doc_id: UUID) -> RAGDocument | None:
"""Get a RAG document by ID."""
return await session.get(RAGDocument, doc_id)
async def create(
session: AsyncSession,
*,
collection_name: str,
filename: str,
filesize: int,
filetype: str,
storage_path: str = "",
) -> RAGDocument:
"""Create a new RAG document record."""
doc = RAGDocument(
collection_name=collection_name,
filename=filename,
filesize=filesize,
filetype=filetype,
storage_path=storage_path,
)
session.add(doc)
await session.flush()
return doc
async def update_status(
session: AsyncSession,
doc_id: UUID,
**kwargs,
) -> RAGDocument | None:
"""Update document status and optional fields."""
stmt = (
update(RAGDocument)
.where(RAGDocument.id == doc_id)
.values(**kwargs)
.returning(RAGDocument)
)
result = await session.execute(stmt)
await session.flush()
return result.scalar_one_or_none()
async def delete(session: AsyncSession, doc_id: UUID) -> None:
"""Delete a RAG document record."""
stmt = delete(RAGDocument).where(RAGDocument.id == doc_id)
await session.execute(stmt)
await session.flush()
async def delete_by_collection(session: AsyncSession, collection_name: str) -> int:
"""Delete all RAG documents for a collection. Returns count."""
stmt = (
delete(RAGDocument)
.where(RAGDocument.collection_name == collection_name)
.returning(RAGDocument.id)
)
result = await session.execute(stmt)
await session.flush()
return len(result.fetchall())
+71
View File
@@ -0,0 +1,71 @@
"""Repository for SyncLog CRUD operations."""
from __future__ import annotations
from datetime import datetime
from uuid import UUID
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession
from backend.db.models.sync_log import SyncLog
async def get_all(
session: AsyncSession,
collection_name: str | None = None,
limit: int = 20,
) -> list[SyncLog]:
"""List sync logs, optionally filtered by collection."""
stmt = (
select(SyncLog)
.order_by(SyncLog.started_at.desc().nulls_last())
.limit(limit)
)
if collection_name:
stmt = stmt.where(SyncLog.collection_name == collection_name)
result = await session.execute(stmt)
return list(result.scalars().all())
async def get_by_id(session: AsyncSession, log_id: UUID) -> SyncLog | None:
"""Get a sync log by ID."""
return await session.get(SyncLog, log_id)
async def create(
session: AsyncSession,
*,
source: str,
collection_name: str,
mode: str,
sync_source_id: UUID | None = None,
) -> SyncLog:
"""Create a new sync log entry."""
log = SyncLog(
source=source,
collection_name=collection_name,
mode=mode,
sync_source_id=sync_source_id,
status="running",
)
session.add(log)
await session.flush()
return log
async def update_status(
session: AsyncSession,
log_id: UUID,
**kwargs,
) -> SyncLog | None:
"""Update sync log status and counters."""
stmt = (
update(SyncLog)
.where(SyncLog.id == log_id)
.values(**kwargs)
.returning(SyncLog)
)
result = await session.execute(stmt)
await session.flush()
return result.scalar_one_or_none()
+100
View File
@@ -0,0 +1,100 @@
"""Repository for SyncSource CRUD operations."""
from __future__ import annotations
from datetime import datetime
from uuid import UUID
from sqlalchemy import delete, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from backend.db.models.sync_source import SyncSource
async def get_all(
session: AsyncSession,
is_active: bool | None = None,
) -> list[SyncSource]:
"""List all sync sources, optionally filtered by active status."""
stmt = select(SyncSource).order_by(SyncSource.created_at.desc())
if is_active is not None:
stmt = stmt.where(SyncSource.is_active == is_active)
result = await session.execute(stmt)
return list(result.scalars().all())
async def get_by_id(session: AsyncSession, source_id: UUID) -> SyncSource | None:
"""Get a sync source by ID."""
return await session.get(SyncSource, source_id)
async def create(
session: AsyncSession,
*,
name: str,
connector_type: str,
collection_name: str,
config: dict | None = None,
sync_mode: str = "full",
schedule_minutes: int | None = None,
) -> SyncSource:
"""Create a new sync source."""
source = SyncSource(
name=name,
connector_type=connector_type,
collection_name=collection_name,
config=config or {},
sync_mode=sync_mode,
schedule_minutes=schedule_minutes,
)
session.add(source)
await session.flush()
return source
async def update(
session: AsyncSession,
source_id: UUID,
**kwargs,
) -> SyncSource | None:
"""Update a sync source."""
stmt = (
update(SyncSource)
.where(SyncSource.id == source_id)
.values(**kwargs)
.returning(SyncSource)
)
result = await session.execute(stmt)
await session.flush()
return result.scalar_one_or_none()
async def delete(session: AsyncSession, source_id: UUID) -> None:
"""Delete a sync source."""
stmt = delete(SyncSource).where(SyncSource.id == source_id)
await session.execute(stmt)
await session.flush()
async def update_sync_status(
session: AsyncSession,
source_id: UUID,
*,
last_sync_at: datetime | None = None,
last_sync_status: str | None = None,
last_error: str | None = None,
) -> SyncSource | None:
"""Update sync source status after a sync operation."""
stmt = (
update(SyncSource)
.where(SyncSource.id == source_id)
.values(
last_sync_at=last_sync_at,
last_sync_status=last_sync_status,
last_error=last_error,
)
.returning(SyncSource)
)
result = await session.execute(stmt)
await session.flush()
return result.scalar_one_or_none()
+487
View File
@@ -0,0 +1,487 @@
"""RAG API routes — collections, search, documents, upload, sync, SSE status.
All endpoints are prefixed with ``/api/rag``.
Authentication is optional for search; admin/owner checks apply on mutations.
"""
from __future__ import annotations
import uuid
from pathlib import Path
from fastapi import APIRouter, Depends, Header, HTTPException, Query, UploadFile
from fastapi.responses import FileResponse, StreamingResponse
from sqlalchemy.ext.asyncio import AsyncSession
from backend.core.config import settings
from backend.core.dependencies import get_db_session
from backend.schemas.rag import (
RAGCollectionInfo,
RAGCollectionList,
RAGDocumentList,
RAGIngestResponse,
RAGMessageResponse,
RAGRetryResponse,
RAGSearchRequest,
RAGSearchResponse,
RAGSyncLogList,
RAGSyncRequest,
RAGSyncResponse,
RAGTrackedDocumentList,
)
from backend.schemas.sync_source import (
ConnectorList,
SyncSourceCreate,
SyncSourceList,
SyncSourceRead,
SyncSourceUpdate,
)
from backend.services.rag.embeddings import EmbeddingService
from backend.services.rag.vectorstore import MilvusVectorStore as VectorStore
from backend.services.rag.retrieval import RetrievalService
from backend.services.rag.ingestion import IngestionService
from backend.services.rag.documents import DocumentProcessor
from backend.services.rag_document import RAGDocumentService
from backend.services.rag_sync import RAGSyncService
from backend.services.rag_status import RAGStatusService
from backend.services.sync_source import SyncSourceService
from backend.rag.connectors import CONNECTOR_REGISTRY
router = APIRouter(prefix="/api/rag", tags=["rag"])
# ── Vector store helpers (lazily initialised) ─────────────────────────────
def _get_vector_store() -> VectorStore:
"""Return a shared VectorStore instance for route handlers."""
rag_settings = settings.rag
embed_service = EmbeddingService(settings=rag_settings)
return VectorStore(settings=rag_settings, embedding_service=embed_service)
def _get_ingestion_service() -> IngestionService:
"""Return a shared IngestionService instance."""
rag_settings = settings.rag
embed_service = EmbeddingService(settings=rag_settings)
vector_store = VectorStore(settings=rag_settings, embedding_service=embed_service)
processor = DocumentProcessor(settings=rag_settings)
return IngestionService(processor=processor, vector_store=vector_store)
# ── Auth helper ───────────────────────────────────────────────────────────
async def _resolve_user_id(
authorization: str | None = Header(None),
) -> str | None:
"""Resolve user from Bearer token (optional for RAG endpoints)."""
if authorization is None:
return None
scheme, _, token = authorization.partition(" ")
if scheme.lower() == "bearer" and token:
from backend.core.database import resolve_auth_token
return await resolve_auth_token(token)
return None
# ═══════════════════════════════════════════════════════════════════════════
# Collection Management
# ═══════════════════════════════════════════════════════════════════════════
@router.get("/collections", response_model=RAGCollectionList)
async def list_collections() -> RAGCollectionList:
"""List all available vector store collections."""
store = _get_vector_store()
names = await store.list_collections()
return RAGCollectionList(items=names)
@router.get("/collections/{name}", response_model=RAGCollectionInfo)
async def get_collection_info(name: str) -> RAGCollectionInfo:
"""Get metadata and stats about a specific collection."""
store = _get_vector_store()
try:
info = await store.get_collection_info(name)
return RAGCollectionInfo(
name=info.name,
total_vectors=info.total_vectors,
dim=info.dim,
indexing_status=info.indexing_status,
)
except Exception as exc:
raise HTTPException(status_code=404, detail=f"Collection not found: {exc}") from exc
@router.post("/collections", response_model=RAGMessageResponse)
async def create_collection(name: str = Query(..., description="Collection name")) -> RAGMessageResponse:
"""Create a new empty collection."""
store = _get_vector_store()
try:
await store.create_collection(name)
return RAGMessageResponse(message=f"Collection '{name}' created")
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
@router.delete("/collections/{name}", response_model=RAGMessageResponse)
async def delete_collection(name: str) -> RAGMessageResponse:
"""Delete a collection and all its data."""
store = _get_vector_store()
await store.delete_collection(name)
return RAGMessageResponse(message=f"Collection '{name}' deleted")
# ═══════════════════════════════════════════════════════════════════════════
# Vector Search
# ═══════════════════════════════════════════════════════════════════════════
@router.post("/search", response_model=RAGSearchResponse)
async def search_rag(
body: RAGSearchRequest,
) -> RAGSearchResponse:
"""Search across one or more collections."""
rag_settings = settings.rag
embed_service = EmbeddingService(settings=rag_settings)
store = VectorStore(settings=rag_settings, embedding_service=embed_service)
retrieval = RetrievalService(vector_store=store, settings=rag_settings)
collection_names = body.collection_names or [body.collection_name]
if len(collection_names) == 1:
results = await retrieval.retrieve(
query=body.query,
collection_name=collection_names[0],
limit=body.limit,
min_score=body.min_score,
filter=body.filter or "",
)
else:
results = await retrieval.retrieve_multi(
query=body.query,
collection_names=collection_names,
limit=body.limit,
min_score=body.min_score,
)
return RAGSearchResponse(
results=[
{
"content": r.content,
"score": r.score,
"metadata": r.metadata,
"parent_doc_id": r.parent_doc_id or "",
}
for r in results
]
)
# ═══════════════════════════════════════════════════════════════════════════
# Document Tracking (SQL-backed)
# ═══════════════════════════════════════════════════════════════════════════
@router.get("/documents", response_model=RAGTrackedDocumentList)
async def list_documents(
collection_name: str | None = Query(None),
session: AsyncSession = Depends(get_db_session),
) -> RAGTrackedDocumentList:
"""List tracked RAG documents, optionally filtered by collection."""
service = RAGDocumentService(db=session)
return await service.list_documents(collection_name=collection_name)
@router.get("/documents/{doc_id}", response_model=RAGTrackedDocumentList)
async def get_document(
doc_id: str,
session: AsyncSession = Depends(get_db_session),
) -> RAGTrackedDocumentList:
"""Get details for a specific tracked document (wrapped in list for consistency)."""
service = RAGDocumentService(db=session)
doc = await service.get_document(doc_id)
from backend.schemas.rag import RAGTrackedDocumentItem, RAGTrackedDocumentList
return RAGTrackedDocumentList(
items=[
RAGTrackedDocumentItem(
id=str(doc.id),
collection_name=doc.collection_name,
filename=doc.filename,
filesize=doc.filesize,
filetype=doc.filetype,
status=doc.status,
error_message=doc.error_message,
vector_document_id=doc.vector_document_id,
chunk_count=doc.chunk_count,
has_file=bool(doc.storage_path),
created_at=doc.created_at.isoformat() if doc.created_at else None,
completed_at=doc.completed_at.isoformat() if doc.completed_at else None,
)
],
total=1,
)
@router.delete("/documents/{doc_id}", response_model=RAGMessageResponse)
async def delete_document(
doc_id: str,
session: AsyncSession = Depends(get_db_session),
) -> RAGMessageResponse:
"""Delete a document from tracking and cascade to vector store + file storage."""
service = RAGDocumentService(db=session)
ingestion = _get_ingestion_service()
await service.delete_document(doc_id, ingestion_service=ingestion)
return RAGMessageResponse(message=f"Document '{doc_id}' deleted")
@router.post("/documents/{doc_id}/retry", response_model=RAGRetryResponse)
async def retry_document(
doc_id: str,
session: AsyncSession = Depends(get_db_session),
) -> RAGRetryResponse:
"""Retry ingestion for a failed document."""
service = RAGDocumentService(db=session)
try:
doc = await service.retry_ingestion(doc_id)
return RAGRetryResponse(
id=str(doc.id),
status=doc.status,
message="Document queued for retry",
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
@router.get("/documents/{doc_id}/download")
async def download_document(
doc_id: str,
session: AsyncSession = Depends(get_db_session),
):
"""Download the original file for a tracked document."""
service = RAGDocumentService(db=session)
try:
file_path, filename, mime_type = await service.get_download_info(doc_id)
return FileResponse(
path=file_path,
filename=filename,
media_type=mime_type,
)
except Exception as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
# ═══════════════════════════════════════════════════════════════════════════
# File Upload & Ingestion
# ═══════════════════════════════════════════════════════════════════════════
@router.post("/upload", response_model=RAGIngestResponse)
@router.post("/upload/{collection_name}", response_model=RAGIngestResponse)
async def upload_document(
file: UploadFile,
collection_name: str = "documents",
replace: bool = Query(True, description="Replace existing document with same name"),
session: AsyncSession = Depends(get_db_session),
):
"""Upload a file for RAG ingestion.
Supported formats: PDF, DOCX, TXT, MD (based on configured parser).
The file is validated, stored, tracked in the database, and queued for
background ingestion via ARQ.
"""
if not file.filename:
raise HTTPException(status_code=400, detail="No filename provided")
file_data = await file.read()
service = RAGDocumentService(db=session)
store = _get_vector_store()
try:
result = await service.dispatch_upload(
collection_name=collection_name,
file_data=file_data,
filename=file.filename,
replace=replace,
vector_store=store,
)
return result
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
# ═══════════════════════════════════════════════════════════════════════════
# Sync Operations
# ═══════════════════════════════════════════════════════════════════════════
@router.get("/sync-logs", response_model=RAGSyncLogList)
async def list_sync_logs(
collection_name: str | None = Query(None),
limit: int = Query(20, ge=1, le=100),
session: AsyncSession = Depends(get_db_session),
) -> RAGSyncLogList:
"""List RAG sync operation logs."""
service = RAGSyncService(db=session)
return await service.list_sync_logs(collection_name=collection_name, limit=limit)
@router.post("/sync", response_model=RAGSyncResponse)
async def trigger_sync(
body: RAGSyncRequest,
session: AsyncSession = Depends(get_db_session),
) -> RAGSyncResponse:
"""Trigger a sync operation for a local path."""
service = RAGSyncService(db=session)
try:
sync_log = await service.start_local_sync(
collection_name=body.collection_name,
mode=body.mode,
path=body.path or None,
)
return RAGSyncResponse(
id=str(sync_log.id),
status=sync_log.status,
message=f"Sync started for collection '{body.collection_name}'",
)
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
@router.post("/sync/{sync_id}/cancel", response_model=RAGMessageResponse)
async def cancel_sync(
sync_id: str,
session: AsyncSession = Depends(get_db_session),
) -> RAGMessageResponse:
"""Cancel a running sync operation."""
service = RAGSyncService(db=session)
try:
await service.cancel_sync(sync_id)
return RAGMessageResponse(message=f"Sync '{sync_id}' cancelled")
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
# ═══════════════════════════════════════════════════════════════════════════
# SSE Status Streaming
# ═══════════════════════════════════════════════════════════════════════════
@router.get("/status")
async def stream_rag_status():
"""SSE endpoint that streams RAG ingestion status events.
Events are consumed from a Redis pub/sub channel ``rag_status``.
Each event has ``event: status`` and JSON data.
"""
status_service = RAGStatusService()
return StreamingResponse(
status_service.stream_events(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
},
)
# ═══════════════════════════════════════════════════════════════════════════
# Connectors
# ═══════════════════════════════════════════════════════════════════════════
@router.get("/connectors", response_model=ConnectorList)
async def list_connectors() -> ConnectorList:
"""List available sync source connector types."""
return SyncSourceService.list_connectors()
# ═══════════════════════════════════════════════════════════════════════════
# Sync Sources (CRUD + trigger)
# ═══════════════════════════════════════════════════════════════════════════
@router.get("/sources", response_model=SyncSourceList)
async def list_sources(
is_active: bool | None = Query(None),
session: AsyncSession = Depends(get_db_session),
) -> SyncSourceList:
"""List sync source configurations."""
service = SyncSourceService(db=session)
return await service.list_sources(is_active=is_active)
@router.get("/sources/{source_id}", response_model=SyncSourceRead)
async def get_source(
source_id: str,
session: AsyncSession = Depends(get_db_session),
) -> SyncSourceRead:
"""Get a specific sync source."""
service = SyncSourceService(db=session)
try:
source = await service.get_source(source_id)
return service._to_read(source) # type: ignore[return-value]
except Exception as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@router.post("/sources", response_model=SyncSourceRead, status_code=201)
async def create_source(
body: SyncSourceCreate,
session: AsyncSession = Depends(get_db_session),
) -> SyncSourceRead:
"""Create a new sync source configuration."""
service = SyncSourceService(db=session)
try:
return await service.create_source(body)
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
@router.patch("/sources/{source_id}", response_model=SyncSourceRead)
async def update_source(
source_id: str,
body: SyncSourceUpdate,
session: AsyncSession = Depends(get_db_session),
) -> SyncSourceRead:
"""Update an existing sync source."""
service = SyncSourceService(db=session)
try:
return await service.update_source(source_id, body)
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
@router.delete("/sources/{source_id}", response_model=RAGMessageResponse)
async def delete_source(
source_id: str,
session: AsyncSession = Depends(get_db_session),
) -> RAGMessageResponse:
"""Delete a sync source."""
service = SyncSourceService(db=session)
try:
await service.delete_source(source_id)
return RAGMessageResponse(message=f"Source '{source_id}' deleted")
except Exception as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@router.post("/sources/{source_id}/sync", response_model=RAGSyncResponse)
async def trigger_source_sync(
source_id: str,
session: AsyncSession = Depends(get_db_session),
) -> RAGSyncResponse:
"""Trigger an immediate sync for a configured source."""
service = SyncSourceService(db=session)
try:
sync_log = await service.trigger_sync(source_id)
return RAGSyncResponse(
id=str(sync_log.id),
status=sync_log.status,
message=f"Sync triggered for source '{source_id}'",
)
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
View File
+158
View File
@@ -0,0 +1,158 @@
"""RAG API schemas."""
from typing import Any
from pydantic import BaseModel, Field
class RAGSearchRequest(BaseModel):
"""Parameters for a vector search query."""
collection_name: str = Field("documents", description="Target collection for search")
collection_names: list[str] | None = Field(
None, description="Search across multiple collections (overrides collection_name)"
)
query: str = Field(..., description="Natural language search query")
limit: int = Field(default=4, ge=1, le=20)
min_score: float = Field(default=0.0, ge=0.0, le=1.0)
filter: str | None = Field(
None, description="Scalar filter expression (e.g. 'filetype == \"pdf\"')"
)
class RAGSearchResult(BaseModel):
"""A single retrieved chunk with its associated metadata."""
content: str
score: float
metadata: dict[str, Any]
parent_doc_id: str
class RAGSearchResponse(BaseModel):
"""List of results found in the vector store."""
results: list[RAGSearchResult]
class RAGCollectionInfo(BaseModel):
"""Statistical information about a specific collection."""
name: str
total_vectors: int
dim: int
indexing_status: str = "complete"
class RAGCollectionList(BaseModel):
"""List of all available collection names."""
items: list[str]
class RAGDocumentItem(BaseModel):
"""Information about a single document in a collection."""
document_id: str = Field(..., description="Unique identifier of the document")
filename: str | None = Field(None, description="Original filename of the document")
filesize: int | None = Field(None, description="Size of the file in bytes")
filetype: str | None = Field(None, description="MIME type of the file")
chunk_count: int = Field(default=0, description="Number of chunks/vectors in the collection")
additional_info: dict[str, Any] | None = Field(None, description="Additional metadata")
class RAGDocumentList(BaseModel):
"""List of all documents in a collection."""
items: list[RAGDocumentItem]
total: int = Field(..., description="Total number of unique documents")
class RAGMessageResponse(BaseModel):
"""Simple message response."""
message: str
class RAGTrackedDocumentItem(BaseModel):
"""A document tracked in the SQL database."""
id: str
collection_name: str
filename: str
filesize: int
filetype: str
status: str
error_message: str | None = None
vector_document_id: str | None = None
chunk_count: int = 0
has_file: bool = False
created_at: str | None = None
completed_at: str | None = None
class RAGTrackedDocumentList(BaseModel):
"""List of tracked RAG documents."""
items: list[RAGTrackedDocumentItem]
total: int
class RAGIngestResponse(BaseModel):
"""Response for document ingestion (async or sync)."""
id: str
status: str
filename: str
collection: str
message: str
document_id: str | None = None
class RAGRetryResponse(BaseModel):
"""Response for document retry."""
id: str
status: str
message: str
class RAGSyncRequest(BaseModel):
"""Request to trigger a sync operation."""
collection_name: str = Field("documents", description="Target collection")
mode: str = Field("full", description="Sync mode: full, new_only, update_only")
path: str = Field("", description="Source path")
class RAGSyncLogItem(BaseModel):
"""A sync operation log entry."""
id: str
source: str
collection_name: str
status: str
mode: str
total_files: int = 0
ingested: int = 0
updated: int = 0
skipped: int = 0
failed: int = 0
error_message: str | None = None
started_at: str | None = None
completed_at: str | None = None
class RAGSyncLogList(BaseModel):
"""List of sync log entries."""
items: list[RAGSyncLogItem]
total: int
class RAGSyncResponse(BaseModel):
"""Response for sync trigger."""
id: str
status: str
message: str
+77
View File
@@ -0,0 +1,77 @@
"""Sync source API schemas."""
from __future__ import annotations
from typing import Any
from pydantic import BaseModel, Field
class ConnectorConfigField(BaseModel):
"""Schema for a single connector configuration field."""
type: str = "string"
description: str = ""
required: bool = False
default: Any = None
class ConnectorInfo(BaseModel):
"""Information about an available connector."""
type: str
name: str
config_schema: dict[str, ConnectorConfigField]
enabled: bool = True
class ConnectorList(BaseModel):
"""List of available connectors."""
items: list[ConnectorInfo]
class SyncSourceCreate(BaseModel):
"""Request to create a sync source."""
name: str = Field(..., description="Human-readable name")
connector_type: str = Field(..., description="Connector type identifier")
collection_name: str = Field("documents", description="Target RAG collection")
config: dict[str, Any] = Field(default_factory=dict)
sync_mode: str = Field("full", description="Sync mode: full, new_only, update_only")
schedule_minutes: int | None = Field(None, description="Auto-sync interval in minutes")
class SyncSourceUpdate(BaseModel):
"""Request to update a sync source."""
name: str | None = Field(None, description="Human-readable name")
collection_name: str | None = Field(None, description="Target RAG collection")
config: dict[str, Any] | None = Field(None)
sync_mode: str | None = Field(None, description="Sync mode")
schedule_minutes: int | None = Field(None, description="Auto-sync interval")
is_active: bool | None = Field(None, description="Whether the source is active")
class SyncSourceRead(BaseModel):
"""Sync source read model returned by the API."""
id: str
name: str
connector_type: str
collection_name: str
config: dict[str, Any] = Field(default_factory=dict)
sync_mode: str = "full"
schedule_minutes: int | None = None
is_active: bool = True
last_sync_at: str | None = None
last_sync_status: str | None = None
last_error: str | None = None
created_at: str | None = None
class SyncSourceList(BaseModel):
"""List of sync sources."""
items: list[SyncSourceRead]
total: int
+141
View File
@@ -0,0 +1,141 @@
"""File storage service for chat file uploads.
Supports local filesystem storage.
Files are organized per-user: {storage_root}/{user_id}/{uuid}_{filename}
"""
import logging
import re
import uuid
from abc import ABC, abstractmethod
from pathlib import Path
from backend.core.config import settings
logger = logging.getLogger(__name__)
ALLOWED_MIME_TYPES = {
"image/jpeg",
"image/png",
"image/gif",
"image/webp",
"text/plain",
"text/markdown",
"text/csv",
"text/html",
"text/css",
"text/xml",
"text/x-python",
"text/javascript",
"text/x-yaml",
"application/json",
"application/pdf",
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
"application/x-yaml",
}
IMAGE_MIME_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"}
MAX_UPLOAD_SIZE = 10 * 1024 * 1024 # 10MB
def classify_file(mime_type: str, filename: str) -> str:
"""Classify file type based on MIME type and extension."""
if mime_type in IMAGE_MIME_TYPES:
return "image"
if mime_type == "application/pdf" or filename.lower().endswith(".pdf"):
return "pdf"
ext = filename.lower().rsplit(".", 1)[-1] if "." in filename else ""
if ext == "docx" or "wordprocessingml" in mime_type:
return "docx"
return "text"
_UNSAFE_FILENAME_CHARS = re.compile(r"[^\w.\-]+")
def _sanitize_filename(filename: str) -> str:
"""Strip path separators, NULL bytes, and unsafe chars from a filename.
The result is always a single path component with no traversal segments.
Empty results fall back to ``"file"`` to preserve a non-empty name.
"""
base = Path(filename).name.replace("\x00", "")
cleaned = _UNSAFE_FILENAME_CHARS.sub("_", base).strip("._")
return cleaned or "file"
def make_storage_filename(filename: str) -> str:
"""Create a unique storage filename to prevent collisions and path traversal."""
safe = _sanitize_filename(filename)
return f"{uuid.uuid4().hex[:12]}_{safe}"
class BaseFileStorage(ABC):
"""Abstract file storage backend."""
@abstractmethod
async def save(self, user_id: str, filename: str, data: bytes) -> str:
"""Save file and return storage path/key."""
@abstractmethod
async def load(self, storage_path: str) -> bytes:
"""Load file bytes by storage path."""
@abstractmethod
async def delete(self, storage_path: str) -> None:
"""Delete file by storage path."""
def get_full_path(self, storage_path: str) -> Path | None:
"""Return absolute filesystem path if available (local storage only)."""
return None # pragma: no cover
class LocalFileStorage(BaseFileStorage):
"""Store files on local filesystem."""
def __init__(self, base_dir: str | Path = "media"):
self.base_dir = Path(base_dir)
self.base_dir.mkdir(parents=True, exist_ok=True)
def _resolve_safe_path(self, storage_path: str) -> Path:
"""Resolve a storage path under base_dir, rejecting traversal attempts."""
base = self.base_dir.resolve()
candidate = (base / storage_path).resolve()
if base != candidate and base not in candidate.parents:
raise ValueError(f"Path escapes storage root: {storage_path}")
return candidate
async def save(self, user_id: str, filename: str, data: bytes) -> str:
safe_user = _sanitize_filename(user_id)
user_dir = self.base_dir / safe_user
user_dir.mkdir(parents=True, exist_ok=True)
storage_name = make_storage_filename(filename)
file_path = user_dir / storage_name
file_path.write_bytes(data)
return f"{safe_user}/{storage_name}"
async def load(self, storage_path: str) -> bytes:
file_path = self._resolve_safe_path(storage_path)
if not file_path.exists():
raise FileNotFoundError(f"File not found: {storage_path}")
return file_path.read_bytes()
async def delete(self, storage_path: str) -> None:
file_path = self._resolve_safe_path(storage_path)
if file_path.exists():
file_path.unlink()
def get_full_path(self, storage_path: str) -> Path | None:
"""Return absolute filesystem path for local files."""
try:
file_path = self._resolve_safe_path(storage_path)
except ValueError:
return None
return file_path if file_path.exists() else None
def get_file_storage() -> BaseFileStorage:
"""Factory: create file storage backend based on settings."""
media_dir = getattr(settings, "media_dir", "media")
return LocalFileStorage(base_dir=media_dir)
+156
View File
@@ -0,0 +1,156 @@
"""File upload service (PostgreSQL async).
Contains business logic for file validation, content parsing, and chat file
creation. Moves parsing helpers and file classification out of the route layer.
"""
import logging
from typing import Any
from sqlalchemy.ext.asyncio import AsyncSession
from backend.db.models.chat_file import ChatFile
from backend.services.file_storage import (
ALLOWED_MIME_TYPES,
MAX_UPLOAD_SIZE,
classify_file,
)
logger = logging.getLogger(__name__)
class FileUploadService:
"""Service for file upload validation, parsing, and persistence."""
ALLOWED_MIME_TYPES = ALLOWED_MIME_TYPES
MAX_UPLOAD_SIZE = MAX_UPLOAD_SIZE
def __init__(self, db: AsyncSession):
self.db = db
@staticmethod
def validate_upload(content_type: str | None, size: int) -> tuple[bool, str | None]:
"""Validate file type and size.
Returns:
Tuple of (is_valid, error_message).
"""
if content_type not in ALLOWED_MIME_TYPES:
return False, f"File type '{content_type}' is not supported."
if size > MAX_UPLOAD_SIZE:
return False, f"File too large. Maximum size is {MAX_UPLOAD_SIZE // (1024 * 1024)}MB."
return True, None
@staticmethod
def classify_file(mime_type: str, filename: str) -> str:
"""Classify file type based on MIME type and extension."""
return classify_file(mime_type, filename)
async def parse_content(
self,
data: bytes,
file_type: str,
mime_type: str = "",
) -> str | None:
"""Parse file content based on file type.
Returns extracted text content or None if parsing fails.
"""
if file_type == "text":
return self._parse_text_content(data, mime_type)
elif file_type == "pdf":
return self._parse_pdf_content(data)
elif file_type == "docx":
return self._parse_docx_content(data)
return None
@staticmethod
def _parse_text_content(data: bytes, mime_type: str) -> str | None:
"""Extract text content from text-based files."""
try:
return data.decode("utf-8")
except (UnicodeDecodeError, ValueError):
return None
@staticmethod
def _parse_pdf_content(data: bytes) -> str | None:
"""Extract text from PDF using PyMuPDF."""
try:
import pymupdf
doc: Any = pymupdf.open(stream=data, filetype="pdf") # type: ignore[no-untyped-call,unused-ignore]
texts = []
for page in doc:
blocks = page.get_text("blocks")
for b in blocks:
if b[6] == 0:
text = b[4].strip()
if text:
texts.append(text)
try:
tables = page.find_tables()
if tables and tables.tables:
for table in tables.tables:
df = table.to_pandas()
if not df.empty:
texts.append(df.to_markdown(index=False))
except Exception:
pass
doc.close()
return "\n\n".join(texts) if texts else None
except Exception as e:
logger.warning(f"PDF parsing failed: {e}")
return None
@staticmethod
def _parse_docx_content(data: bytes) -> str | None:
"""Extract text from DOCX."""
try:
import io
from docx import Document as DOCXDocument
doc: Any = DOCXDocument(io.BytesIO(data))
return "\n".join(p.text for p in doc.paragraphs if p.text.strip())
except Exception as e:
logger.warning(f"DOCX parsing failed: {e}")
return None
async def get_user_file(self, file_id: Any, user_id: Any) -> ChatFile:
"""Get a file by ID, verifying ownership.
Raises:
NotFoundError: If file does not exist or user has no access.
"""
from backend.core.exceptions import NotFoundError
from backend.repositories import chat_file_repo
chat_file = await chat_file_repo.get_by_id(self.db, file_id)
if not chat_file or str(chat_file.user_id) != str(user_id):
raise NotFoundError(message="File not found")
return chat_file
async def create_chat_file(
self,
*,
user_id: Any,
filename: str,
mime_type: str,
size: int,
storage_path: str,
file_type: str,
parsed_content: str | None = None,
) -> ChatFile:
"""Create a chat file record in the database."""
from backend.repositories import chat_file_repo
return await chat_file_repo.create(
self.db,
user_id=user_id,
filename=filename,
mime_type=mime_type,
size=size,
storage_path=storage_path,
file_type=file_type,
parsed_content=parsed_content,
)
+164
View File
@@ -0,0 +1,164 @@
"""RAG configuration."""
from enum import StrEnum
from pydantic import BaseModel, Field, model_validator
class DocumentExtensions(StrEnum):
"""Extensions supported by the RAG ingestion pipeline."""
PDF = ".pdf"
DOCX = ".docx"
MD = ".md"
TXT = ".txt"
# Supported file formats per parser
PYMUPDF_FORMATS: set[str] = {".pdf", ".docx", ".txt", ".md"}
LITEPARSE_FORMATS: set[str] = {
".pdf",
".docx",
".xlsx",
".pptx",
".txt",
".md",
".jpg",
".jpeg",
".png",
".tiff",
}
LLAMAPARSE_FORMATS: set[str] = {
# Documents
".pdf",
".docx",
".doc",
".pptx",
".ppt",
".rtf",
".txt",
".md",
".epub",
# Images
".jpg",
".jpeg",
".png",
".gif",
".bmp",
".tiff",
".webp",
# Spreadsheets
".xlsx",
".xls",
".csv",
".tsv",
".ods",
# Audio
".mp3",
".mp4",
".wav",
".m4a",
".webm",
# Web
".html",
".htm",
".xml",
}
PARSER_FORMATS: dict[str, set[str]] = {
"pymupdf": PYMUPDF_FORMATS,
"liteparse": LITEPARSE_FORMATS,
"llamaparse": LLAMAPARSE_FORMATS,
}
def get_supported_formats(parser_name: str = "pymupdf") -> set[str]:
"""Get supported file formats for a given parser."""
return PARSER_FORMATS.get(parser_name, PYMUPDF_FORMATS)
# Known embedding models and their output dimensions.
# Used to auto-set vector store dimension from model name.
EMBEDDING_DIMENSIONS: dict[str, int] = {
# OpenAI
"text-embedding-3-small": 1536,
"text-embedding-3-large": 3072,
"text-embedding-ada-002": 1536,
# Voyage AI
"voyage-3": 1024,
"voyage-3-lite": 512,
"voyage-code-3": 1024,
# Google Gemini
"gemini-embedding-exp-03-07": 3072,
# SentenceTransformers (local)
"all-MiniLM-L6-v2": 384,
"all-mpnet-base-v2": 768,
"bge-small-en-v1.5": 384,
"bge-base-en-v1.5": 768,
"bge-large-en-v1.5": 1024,
}
class EmbeddingsConfig(BaseModel):
"""Embeddings configuration. Dimension is auto-derived from model name."""
model: str = "text-embedding-3-small"
dim: int = 1536
@model_validator(mode="after")
def set_dim_from_model(self) -> "EmbeddingsConfig":
if self.model in EMBEDDING_DIMENSIONS:
self.dim = EMBEDDING_DIMENSIONS[self.model]
return self
class RerankerConfig(BaseModel):
"""Reranker configuration."""
model: str = "cross_encoder"
class DocumentParser(BaseModel):
"""Document parsing settings (non-PDF files)."""
method: str = "python_native"
class PdfParser(BaseModel):
"""PDF parsing settings."""
method: str = "pymupdf"
class RAGSettings(BaseModel):
"""RAG pipeline configuration."""
collection_name: str = "documents"
allowed_extensions: list[DocumentExtensions] = Field(
default_factory=lambda: list(DocumentExtensions)
)
# Chunking
chunk_size: int = 512
chunk_overlap: int = 50
chunking_strategy: str = "recursive"
enable_hybrid_search: bool = False
enable_ocr: bool = False
# Embeddings
embeddings_config: EmbeddingsConfig = Field(default_factory=EmbeddingsConfig)
# Reranker
reranker_config: RerankerConfig = Field(default_factory=RerankerConfig)
# Parsers
document_parser: DocumentParser = Field(default_factory=DocumentParser)
pdf_parser: PdfParser = Field(default_factory=PdfParser)
# Image description
enable_image_description: bool = True
image_description_model: str = ""
# Sources
gdrive_ingestion: bool = False
+467
View File
@@ -0,0 +1,467 @@
import logging
from abc import ABC, abstractmethod
from pathlib import Path
from typing import Any
import pymupdf
from docx import Document as DOCXDocument
from langchain_text_splitters import RecursiveCharacterTextSplitter
from backend.services.rag.config import DocumentExtensions, RAGSettings
from backend.services.rag.models import (
Document,
DocumentImage,
DocumentMetadata,
DocumentPage,
DocumentPageChunk,
)
logger = logging.getLogger(__name__)
class BaseDocumentParser(ABC):
"""Abstract base class for document parsing strategies.
Defines the interface that all document parsers must implement.
Supports parsing of various document formats (PDF, DOCX, TXT, MD).
"""
allowed = [f"{ext.value}" for ext in DocumentExtensions]
def is_file_existing(self, filepath: Path) -> bool:
"""Check if file exists at the given path.
Args:
filepath: Path to the file to check.
Returns:
True if the file exists, False otherwise.
"""
return Path.exists(filepath)
def is_extension_allowed(self, filepath: Path) -> bool:
"""Check whether document extension is allowed for parsing.
Args:
filepath: Path to the file to check.
Returns:
True if the extension is supported and file exists.
"""
return filepath.suffix.lower() in self.allowed and self.is_file_existing(filepath)
def get_document_metadata(self, filepath: Path) -> DocumentMetadata:
"""Collect metadata about a given document.
Args:
filepath: Path to the document file.
Returns:
DocumentMetadata object containing file information.
"""
import hashlib
content_hash = hashlib.sha256(filepath.read_bytes()).hexdigest()
return DocumentMetadata(
filename=filepath.name,
filesize=filepath.stat().st_size,
filetype=filepath.suffix.replace(".", ""),
source_path=str(filepath.resolve()),
content_hash=content_hash,
)
@abstractmethod
async def parse(self, filepath: Path) -> Document:
"""Parse a file and read its content into a Document object.
Args:
filepath: Path to the file to parse.
Returns:
Document object with parsed content and metadata.
"""
pass
class TextDocumentParser(BaseDocumentParser):
"""Parser for text-based documents (TXT, MD).
Uses Python's built-in file reading capabilities to extract
text content from plain text and Markdown files.
"""
def _parse_text_file(self, filepath: Path) -> Document:
"""Extract raw text from a TXT or MD file.
Args:
filepath: Path to the text file.
Returns:
Document object with the file content.
"""
with open(filepath, encoding="utf-8") as f:
page = DocumentPage(page_num=1, content=f.read())
return Document(pages=[page], metadata=self.get_document_metadata(filepath))
async def parse(self, filepath: Path) -> Document:
"""Parse a text file (TXT or MD).
Args:
filepath: Path to the text file.
Returns:
Document object with parsed content.
Raises:
ValueError: If the file extension is not supported.
"""
if not self.is_extension_allowed(filepath):
raise ValueError(f"Extension {filepath.suffix} not supported by TextDocumentParser")
if filepath.suffix in (".txt", ".md"):
return self._parse_text_file(filepath)
else:
raise ValueError(f"Unsupported file extension. Allowed extensions: {self.allowed}")
class DocxDocumentParser(BaseDocumentParser):
"""Parser for DOCX documents using python-docx.
Extracts text content from Microsoft Word documents by reading
all paragraphs and joining them with newline characters.
"""
def _parse_docx_file(self, filepath: Path) -> Document:
"""Extract raw text from the DOCX file.
Args:
filepath: Path to the DOCX file.
Returns:
Document object with the file content.
"""
file: Any = DOCXDocument(str(filepath))
page = DocumentPage(page_num=1, content="\n".join([p.text for p in file.paragraphs]))
return Document(pages=[page], metadata=self.get_document_metadata(filepath))
async def parse(self, filepath: Path) -> Document:
"""Parse a DOCX file.
Args:
filepath: Path to the DOCX file.
Returns:
Document object with parsed content.
Raises:
ValueError: If the file is not a DOCX file.
"""
if not self.is_extension_allowed(filepath):
raise ValueError(f"Extension {filepath.suffix} not supported by DocxDocumentParser")
if filepath.suffix == ".docx":
return self._parse_docx_file(filepath)
else:
raise ValueError(f"Unsupported file extension. Allowed extensions: {self.allowed}")
class PyMuPDFParser(BaseDocumentParser):
"""Smart PDF parser using PyMuPDF.
Features:
- Text extraction with layout preservation (blocks)
- Table detection → markdown tables
- Header/footer detection and removal
- OCR fallback for scanned pages (optional, requires tesseract)
- Image extraction for LLM-based description
- Document metadata (author, title, TOC)
"""
MIN_TEXT_LENGTH = 50 # below this → likely a scan, try OCR
def __init__(self, enable_ocr: bool = False, image_describer: Any = None):
self.enable_ocr = enable_ocr
self._image_describer = image_describer
def _detect_repeated_content(self, doc: Any) -> set[str]:
"""Detect headers/footers — text appearing on >70% of pages."""
if len(doc) < 3:
return set()
text_counts: dict[str, int] = {}
for page in doc:
for b in page.get_text("blocks"):
if b[6] != 0: # skip image blocks
continue
y_ratio = b[1] / page.rect.height if page.rect.height else 0
if y_ratio < 0.15 or y_ratio > 0.85:
text = b[4].strip()
if text and len(text) < 200:
text_counts[text] = text_counts.get(text, 0) + 1
threshold = len(doc) * 0.7
return {t for t, c in text_counts.items() if c >= threshold}
def _extract_text(self, page: Any, repeated: set[str]) -> str:
"""Extract text blocks, filtering headers/footers."""
texts = []
for b in page.get_text("blocks"):
if b[6] != 0: # skip image blocks
continue
text = b[4].strip()
if text and text not in repeated:
texts.append(text)
return str("\n\n".join(texts))
def _extract_tables(self, page: Any) -> str:
"""Extract tables as markdown."""
try:
tables = page.find_tables()
if not tables or not tables.tables:
return ""
parts = []
for table in tables.tables:
df = table.to_pandas()
if not df.empty:
parts.append(df.to_markdown(index=False))
return "\n\n".join(parts)
except Exception:
return ""
def _ocr_page(self, page: Any, image_describer: Any = None) -> str:
"""OCR a scanned page by rendering it as image and sending to LLM vision."""
if not image_describer:
return ""
try:
import asyncio
pix = page.get_pixmap(dpi=200)
image_bytes = pix.tobytes("png")
loop = asyncio.new_event_loop()
try:
return str(
loop.run_until_complete(image_describer.describe(image_bytes, "image/png"))
)
finally:
loop.close()
except Exception as e:
logger.warning(f"LLM OCR failed for page {page.number + 1}: {e}")
return ""
def _extract_images(self, doc: Any, page: Any) -> list["DocumentImage"]:
"""Extract images from page for LLM description."""
images = []
for img_info in page.get_images(full=True):
xref = img_info[0]
try:
base = doc.extract_image(xref)
if base and base["image"] and len(base["image"]) > 1000:
ext = base.get("ext", "png")
mime_map = {"png": "image/png", "jpeg": "image/jpeg", "jpg": "image/jpeg"}
images.append(
DocumentImage(
page_num=page.number + 1,
image_bytes=base["image"],
mime_type=mime_map.get(ext, f"image/{ext}"),
)
)
except Exception:
pass
return images
def _parse_pdf_file(self, filepath: Path) -> Document:
"""Parse PDF with smart extraction pipeline."""
doc: Any = pymupdf.open(filepath) # type: ignore[no-untyped-call]
# Doc-level metadata
meta = doc.metadata or {}
toc = doc.get_toc()
# Detect repeated headers/footers
repeated = self._detect_repeated_content(doc)
pages = []
for page in doc:
# 1. Text with layout (skip image blocks, filter headers/footers)
text = self._extract_text(page, repeated)
# 2. Tables → markdown
tables_md = self._extract_tables(page)
if tables_md:
text = text + "\n\n" + tables_md if text.strip() else tables_md
# 3. OCR fallback for scans/empty pages
if self.enable_ocr and len(text.strip()) < self.MIN_TEXT_LENGTH:
ocr_text = self._ocr_page(page, self._image_describer)
if len(ocr_text.strip()) > len(text.strip()):
text = ocr_text
logger.info(f"OCR fallback used for page {page.number + 1}")
# 4. Images
images = self._extract_images(doc, page)
pages.append(
DocumentPage(
page_num=page.number + 1,
content=text,
images=images,
)
)
doc.close()
# Enrich metadata
additional: dict[str, Any] = {}
if meta.get("title"):
additional["pdf_title"] = meta["title"]
if meta.get("author"):
additional["pdf_author"] = meta["author"]
if toc:
additional["toc"] = [{"level": t[0], "title": t[1], "page": t[2]} for t in toc[:20]]
doc_meta = self.get_document_metadata(filepath)
if additional:
doc_meta.additional_info = {**(doc_meta.additional_info or {}), **additional}
return Document(pages=pages, metadata=doc_meta)
async def parse(self, filepath: Path) -> Document:
if not self.is_extension_allowed(filepath):
raise ValueError(f"Extension {filepath.suffix} not supported by PyMuPDFParser")
if filepath.suffix == ".pdf":
return self._parse_pdf_file(filepath)
raise ValueError(f"Unsupported: {filepath.suffix}")
class DocumentProcessor:
"""Orchestrates parsing and chunking of files into Document objects.
Manages the document processing pipeline:
1. Route to appropriate parser based on file extension
2. Parse document content
3. Chunk document pages using RecursiveCharacterTextSplitter
Supported file types:
- TXT, MD: TextDocumentParser (Python native)
- DOCX: DocxDocumentParser (Python native)
- PDF: PyMuPDFParser (local, tables, OCR fallback)
"""
def __init__(self, settings: RAGSettings):
"""Initialize the document processor.
Args:
settings: RAG configuration settings.
"""
self.settings = settings
self.splitter = self._create_splitter(settings)
# Always use Python native parser for plain text
self.text_parser = TextDocumentParser()
self.docx_parser = DocxDocumentParser()
# Image describer for LLM-based image descriptions and OCR fallback
self.image_describer = (
self._init_image_describer(settings) if settings.enable_image_description else None
)
self.pdf_parser = PyMuPDFParser(
enable_ocr=settings.enable_ocr,
image_describer=self.image_describer,
)
@staticmethod
def _create_splitter(settings: RAGSettings) -> Any:
"""Create text splitter based on chunking strategy."""
from langchain_text_splitters import (
MarkdownHeaderTextSplitter,
)
strategy = settings.chunking_strategy
if strategy == "markdown":
# Split by markdown headers, then by size
return MarkdownHeaderTextSplitter(
headers_to_split_on=[
("#", "h1"),
("##", "h2"),
("###", "h3"),
],
strip_headers=False,
)
if strategy == "fixed":
# Simple fixed-size chunks with no smart splitting
return RecursiveCharacterTextSplitter(
chunk_size=settings.chunk_size,
chunk_overlap=settings.chunk_overlap,
length_function=len,
separators=["\n"],
)
# Default: recursive (smart splitting by paragraphs, sentences, words)
return RecursiveCharacterTextSplitter(
chunk_size=settings.chunk_size,
chunk_overlap=settings.chunk_overlap,
length_function=len,
is_separator_regex=False,
)
@staticmethod
def _init_image_describer(settings: RAGSettings) -> Any:
"""Initialize the image describer using the configured AI framework."""
from backend.core.config import settings as app_settings
model_name = (
getattr(app_settings, "rag_image_description_model", None) or app_settings.ai_model
)
from backend.services.rag.image_describer import PydanticAIImageDescriber
return PydanticAIImageDescriber(model_name=model_name)
async def _describe_images(self, document: Document) -> None:
"""Generate text descriptions for all images in document pages."""
for page in document.pages:
if not page.images:
continue
for image in page.images:
image.description = await self.image_describer.describe(
image.image_bytes, image.mime_type
)
img_descriptions = [
f"[Image: {img.description}]" for img in page.images if img.description
]
if img_descriptions:
page.content = f"{page.content}\n\n{chr(10).join(img_descriptions)}"
async def process_file(self, filepath: Path) -> Document:
"""Main entry point: filepath -> Document with chunks.
Args:
filepath: Path to the file to process.
Returns:
Document object with parsed pages and chunked content.
Raises:
ValueError: If the file type is not supported.
"""
# Route to appropriate parser based on file extension
if filepath.suffix in (".txt", ".md"):
document = await self.text_parser.parse(filepath)
elif filepath.suffix == ".docx":
document = await self.docx_parser.parse(filepath)
elif filepath.suffix == ".pdf":
document = await self.pdf_parser.parse(filepath)
else:
raise ValueError(f"Unsupported file type: {filepath.suffix}")
# Describe images using LLM vision before chunking
await self._describe_images(document)
pages = document.pages
chunked_pages: list[DocumentPageChunk] = []
is_markdown_splitter = self.settings.chunking_strategy == "markdown"
for page in pages:
if is_markdown_splitter:
# MarkdownHeaderTextSplitter returns Document objects
md_docs = self.splitter.split_text(page.content)
chunks = [doc.page_content for doc in md_docs]
else:
chunks = self.splitter.split_text(page.content)
for chunk_num, chunk in enumerate(chunks):
chunked_pages.append(
DocumentPageChunk(
chunk_content=chunk,
chunk_num=chunk_num,
parent_doc_id=document.id,
**page.model_dump(exclude={"parent_doc_id"}),
)
)
# Add chunked pages to original document
document.chunked_pages = chunked_pages
return document
+153
View File
@@ -0,0 +1,153 @@
from abc import ABC, abstractmethod
from openai import OpenAI
from backend.services.rag.config import RAGSettings
from backend.services.rag.models import Document
class BaseEmbeddingProvider(ABC):
"""Abstract base class for embedding providers.
Defines the interface that all embedding providers must implement.
"""
@abstractmethod
def embed_queries(self, texts: list[str]) -> list[list[float]]:
"""Embed a list of query texts.
Args:
texts: List of text strings to embed.
Returns:
List of embedding vectors, one for each input text.
"""
pass
@abstractmethod
def embed_document(self, document: Document) -> list[list[float]]:
"""Embed all chunks of a document.
Args:
document: Document object containing chunked pages to embed.
Returns:
List of embedding vectors, one for each chunk in the document.
"""
pass
@abstractmethod
def warmup(self) -> None:
"""Ensures the model is loaded and ready for inference."""
pass
class OpenAIEmbeddingProvider(BaseEmbeddingProvider):
"""OpenAI embedding provider using the OpenAI API.
Uses OpenAI's embedding models to generate text embeddings.
"""
def __init__(self, model: str, base_url: str | None = None) -> None:
"""Initialize the OpenAI embedding provider.
Args:
model: The OpenAI embedding model name (e.g., 'text-embedding-3-small').
base_url: Explicit base URL for the OpenAI API. If not provided,
defaults to the standard OpenAI API, explicitly ignoring any
OPENAI_BASE_URL env var (which may point to a chat-only local
server that doesn't support embeddings).
"""
self.model = model
# Explicitly pass base_url to avoid picking up OPENAI_BASE_URL from env
self.client = OpenAI(base_url=base_url or "https://api.openai.com/v1")
def embed_queries(self, texts: list[str]) -> list[list[float]]:
"""Embed a list of query texts using OpenAI.
Args:
texts: List of text strings to embed.
Returns:
List of embedding vectors.
"""
response = self.client.embeddings.create(model=self.model, input=texts)
return [data.embedding for data in response.data]
def embed_document(self, document: Document) -> list[list[float]]:
"""Embed all chunks of a document using OpenAI.
Args:
document: Document object containing chunked pages.
Returns:
List of embedding vectors for each chunk.
"""
texts = [
doc.chunk_content if doc.chunk_content else "" for doc in (document.chunked_pages or [])
]
return self.embed_queries(texts)
def warmup(self) -> None:
"""Warmup method for OpenAI client.
OpenAI API is a remote service, so this is a no-op.
"""
pass
# Embedding orchestrator
class EmbeddingService:
"""Service for managing text embeddings.
Orchestrates embedding operations using a configured embedding provider.
Supports multiple backends: OpenAI, Voyage AI, and Sentence Transformers.
"""
def __init__(self, settings: RAGSettings):
"""Initialize the embedding service.
Args:
settings: RAG configuration settings.
"""
config = settings.embeddings_config
self.expected_dim = config.dim
self.provider = OpenAIEmbeddingProvider(model=config.model)
def embed_query(self, query: str) -> list[float]:
"""Embed a single query text.
Args:
query: The text query to embed.
Returns:
Embedding vector for the query.
"""
result = self.provider.embed_queries([query])[0]
if len(result) != self.expected_dim:
raise ValueError(
f"Embedding dimension mismatch: expected {self.expected_dim}, "
f"got {len(result)}. Check your embedding model configuration."
)
return result
def embed_document(self, document: Document) -> list[list[float]]:
"""Embed all chunks of a document.
Args:
document: Document object containing chunked pages.
Returns:
List of embedding vectors for each chunk.
"""
results = self.provider.embed_document(document)
if results and len(results[0]) != self.expected_dim:
raise ValueError(
f"Embedding dimension mismatch: expected {self.expected_dim}, "
f"got {len(results[0])}. Check your embedding model configuration."
)
return results
def warmup(self) -> None:
"""Ensures the provider is ready for usage."""
self.provider.warmup()
+64
View File
@@ -0,0 +1,64 @@
"""LLM-based image description for RAG document processing.
Uses the configured AI framework (PydanticAI, LangChain, etc.) to describe
images extracted from documents. Descriptions are appended to page content
before chunking, making image content searchable via text embeddings.
Configuration:
RAG_IMAGE_DESCRIPTION_MODEL — LLM model to use (defaults to AI_MODEL from .env)
"""
import base64
import logging
from abc import ABC, abstractmethod
logger = logging.getLogger(__name__)
IMAGE_DESCRIPTION_PROMPT = (
"Describe this image in detail. Focus on any text, data, charts, diagrams, "
"or visual information that would be useful for document search and retrieval. "
"Be concise but comprehensive."
)
class BaseImageDescriber(ABC):
"""Abstract base for LLM-based image description."""
@abstractmethod
async def describe(self, image_bytes: bytes, mime_type: str = "image/png") -> str:
"""Generate a text description of an image."""
def _b64_encode(image_bytes: bytes) -> str:
"""Base64-encode raw image bytes."""
return base64.b64encode(image_bytes).decode("utf-8")
class PydanticAIImageDescriber(BaseImageDescriber):
"""Image description using PydanticAI (supports all providers)."""
def __init__(self, model_name: str | None = None):
from backend.core.config import settings
self.model_name = (
model_name
or getattr(settings, "rag_image_description_model", None)
or settings.ai_model
)
async def describe(self, image_bytes: bytes, mime_type: str = "image/png") -> str:
try:
from pydantic_ai import Agent
from pydantic_ai.messages import BinaryContent
agent = Agent(self.model_name)
result = await agent.run(
[
BinaryContent(data=image_bytes, media_type=mime_type),
IMAGE_DESCRIPTION_PROMPT,
]
)
return result.output if hasattr(result, "output") else str(result.data)
except Exception as e:
logger.error(f"PydanticAI image description failed: {e}")
return ""
+199
View File
@@ -0,0 +1,199 @@
from __future__ import annotations
import logging
from collections.abc import Awaitable, Callable
from pathlib import Path
from backend.services.rag.documents import DocumentProcessor
from backend.services.rag.models import Document, IngestionResult, IngestionStatus
from backend.services.rag.vectorstore import BaseVectorStore
logger = logging.getLogger(__name__)
class IngestionService:
"""
Orchestrates the data flow:
File Path -> Parse/Chunk -> Deduplicate -> Embed/Store -> Query-Ready
"""
def __init__(
self,
processor: DocumentProcessor,
vector_store: BaseVectorStore,
on_event: Callable[..., Awaitable[None]] | None = None,
):
self.processor = processor
self.store = vector_store
self._on_event = on_event
@classmethod
def from_settings(
cls,
on_event: Callable[..., Awaitable[None]] | None = None,
) -> IngestionService:
"""Build an IngestionService using the application's RAG settings."""
from backend.core.config import settings
from backend.services.rag.embeddings import EmbeddingService
from backend.services.rag.vectorstore import MilvusVectorStore as VectorStore
rag_settings = settings.rag
embed_service = EmbeddingService(settings=rag_settings)
vector_store = VectorStore(settings=rag_settings, embedding_service=embed_service)
processor = DocumentProcessor(settings=rag_settings)
return cls(processor=processor, vector_store=vector_store, on_event=on_event)
async def _emit(self, event: str, data: dict[str, object]) -> None:
"""Emit a webhook event if callback is configured."""
if self._on_event:
try:
await self._on_event(event, data)
except Exception as e:
logger.warning(f"Webhook event dispatch failed: {e}")
async def _find_existing_by_source(self, collection_name: str, source_path: str) -> str | None:
"""Find an existing document by source_path.
Returns the document_id if found, None otherwise.
"""
try:
docs = await self.store.get_documents(collection_name)
for doc in docs:
meta = doc.additional_info or {}
if meta.get("source_path") == source_path:
return doc.document_id
# Also check top-level metadata fields
# (source_path is stored in metadata dict per chunk)
# Fallback: check filename match
for doc in docs:
if doc.filename and doc.filename == Path(source_path).name:
return doc.document_id
except Exception:
pass
return None
async def _find_existing_by_hash(self, collection_name: str, content_hash: str) -> str | None:
"""Find an existing document by content hash (exact duplicate check)."""
try:
docs = await self.store.get_documents(collection_name)
for doc in docs:
meta = doc.additional_info or {}
if meta.get("content_hash") == content_hash:
return doc.document_id
except Exception:
pass
return None
async def ingest_file(
self,
filepath: Path,
collection_name: str,
replace: bool = True,
source_path: str = "",
) -> IngestionResult:
"""Processes a file and pushes it into the vector database.
Args:
filepath: Path to the file to process.
collection_name: Target collection name.
replace: If True, replace existing document with same source_path.
source_path: Override source path (e.g., gdrive://id, s3://bucket/key).
"""
try:
# Processing (Parsing + Chunking)
document: Document = await self.processor.process_file(filepath)
# Set source_path override if provided (e.g., from GDrive/S3)
if source_path:
document.metadata.source_path = source_path
document.metadata.filename = Path(source_path).name
# Deduplication check
existing_id = None
if replace:
# Check by source_path first
if document.metadata.source_path:
existing_id = await self._find_existing_by_source(
collection_name, document.metadata.source_path
)
# If not found by path, check by content hash (exact duplicate)
if not existing_id and document.metadata.content_hash:
existing_id = await self._find_existing_by_hash(
collection_name, document.metadata.content_hash
)
if existing_id:
# Remove old version before inserting new
await self.store.delete_document(collection_name, existing_id)
logger.info(f"Replaced existing document {existing_id} for '{filepath.name}'")
# Storage (Embedding + Insertion)
await self.store.insert_document(
collection_name=collection_name,
document=document,
)
action = "replaced" if existing_id else "ingested"
await self._emit(
"rag.document.ingested",
{
"document_id": document.id,
"filename": filepath.name,
"collection": collection_name,
"action": action,
"chunks": len(document.chunked_pages or []),
"source_path": document.metadata.source_path,
},
)
return IngestionResult(
status=IngestionStatus.DONE,
document_id=document.id,
message=f"Successfully {action} '{filepath.name}'",
)
except Exception as e:
logger.error(f"Ingestion error for {filepath.name}: {e!s}")
return IngestionResult(
status=IngestionStatus.ERROR,
error_message=str(e),
message=f"Failed to process {filepath.name}",
)
async def find_existing(self, collection_name: str, source_path: str) -> str | None:
"""Check if a document with this source_path already exists. Returns document_id or None."""
return await self._find_existing_by_source(collection_name, source_path)
async def get_existing_hash(self, collection_name: str, source_path: str) -> str | None:
"""Get content_hash of existing document by source_path."""
doc_id = await self._find_existing_by_source(collection_name, source_path)
if not doc_id:
return None
try:
docs = await self.store.get_documents(collection_name)
for doc in docs:
if doc.document_id == doc_id and doc.additional_info:
return doc.additional_info.get("content_hash")
except Exception:
pass
return None
async def remove_document(self, collection_name: str, document_id: str) -> bool:
"""Wipes all traces of a document from the vector store."""
try:
await self.store.delete_document(
collection_name=collection_name,
document_id=document_id,
)
await self._emit(
"rag.document.deleted",
{
"document_id": document_id,
"collection": collection_name,
},
)
return True
except Exception as e:
logger.error(f"Failed to delete document {document_id}: {e!s}")
return False
+116
View File
@@ -0,0 +1,116 @@
"""RAG Data Models.
Structures used to interface with the RAG feature."""
import uuid
from enum import StrEnum
from typing import Any
from pydantic import BaseModel, Field, computed_field, model_validator
class DocumentImage(BaseModel):
"""An image extracted from a document page."""
image_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
page_num: int = 0
image_bytes: bytes = b""
description: str = ""
mime_type: str = "image/png"
class DocumentPage(BaseModel):
"""Content of document's page."""
page_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
page_num: int
content: str
parent_doc_id: str | None = None
images: list[DocumentImage] = Field(default_factory=list)
class DocumentPageChunk(DocumentPage):
"""Content of chunked document's page."""
chunk_content: str
chunk_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
chunk_num: int = 0
class DocumentMetadata(BaseModel):
"""Metadata of a document."""
filename: str
filesize: int
filetype: str
source_path: str = "" # original path: local path, s3://bucket/key, gdrive://file_id
content_hash: str = "" # SHA256 hash for deduplication
additional_info: dict[str, Any] | None = None
class Document(BaseModel):
"""A Document object that describes an ingested file."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
pages: list[DocumentPage]
chunked_pages: list[DocumentPageChunk] | None = None
metadata: DocumentMetadata
@computed_field # type: ignore[prop-decorator]
@property
def num_pages(self) -> int:
return len(self.pages)
@model_validator(mode="after")
def connect_pages(self) -> "Document":
for page in self.pages:
page.parent_doc_id = self.id
return self
class SearchResult(BaseModel):
"""A schema of vector store query output."""
content: str
score: float
metadata: dict[str, Any] = Field(default_factory=dict)
parent_doc_id: str | None = None
class IngestionStatus(StrEnum):
"""A collection of available ingestion statuses."""
NEW = "new"
PROCESSING = "processing"
ADDING = "adding"
DONE = "done"
ERROR = "error"
class IngestionResult(BaseModel):
"""A schema to handle document ingestion results."""
status: IngestionStatus = IngestionStatus.NEW
message: str | None = None
error_message: str | None = None
document_id: str | None = None
class CollectionInfo(BaseModel):
"""Collection of information about given collection."""
name: str
total_vectors: int
dim: int
indexing_status: str = "complete"
class DocumentInfo(BaseModel):
"""Information about a document stored in a collection."""
document_id: str
filename: str | None = None
filesize: int | None = None
filetype: str | None = None
chunk_count: int = 0
additional_info: dict[str, Any] | None = None
+271
View File
@@ -0,0 +1,271 @@
"""Reranker implementations for improving RAG retrieval quality.
This module provides reranking functionality to improve the relevance of
search results. It supports both API-based rerankers (Cohere) and local
models (Cross Encoder).
"""
import logging
import time
from abc import ABC, abstractmethod
from backend.core.config import settings
from backend.services.rag.config import RAGSettings
from backend.services.rag.models import SearchResult
logger = logging.getLogger(__name__)
class BaseReranker(ABC):
"""Abstract base class for reranking implementations.
Defines the interface that all reranker providers must implement.
Rerankers take an initial set of search results and reorder them
based on semantic relevance to the query.
"""
@abstractmethod
async def rerank(
self,
query: str,
results: list[SearchResult],
top_k: int,
) -> list[SearchResult]:
"""Rerank search results based on query relevance.
Args:
query: The original search query.
results: Initial search results from vector search.
top_k: Number of top results to return after reranking.
Returns:
Reranked list of SearchResult objects, sorted by relevance.
"""
pass
@abstractmethod
def warmup(self) -> None:
"""Ensure the reranker model is loaded and ready for inference.
For API-based rerankers, this may validate credentials.
For local models, this triggers model download and loading.
"""
pass
@property
@abstractmethod
def name(self) -> str:
"""Return the name of the reranker for logging purposes."""
pass
from sentence_transformers import CrossEncoder
class CrossEncoderReranker(BaseReranker):
"""Cross Encoder reranker using local Sentence Transformers model.
Uses a cross-encoder model to score query-document pairs for relevance.
Runs entirely locally - no API calls required.
Default model: cross-encoder/ms-marco-MiniLM-L6-v2 (lightweight, fast)
"""
# Default cross-encoder model for reranking
DEFAULT_MODEL = settings.cross_encoder_model
def __init__(self, model: str | None = None, cache_dir: str | None = None):
"""Initialize the Cross Encoder reranker.
Args:
model: Cross-encoder model name from Sentence Transformers.
Defaults to cross-encoder/ms-marco-MiniLM-L6-v2 if not specified.
cache_dir: Directory to cache the model. Defaults to app models cache.
"""
self.model_name = model or self.DEFAULT_MODEL
self.cache_dir = cache_dir
self._model = None
@property
def model(self) -> CrossEncoder:
"""Lazy load the cross-encoder model."""
if self._model is None:
from backend.core.config import settings as app_settings
cache_path = self.cache_dir or str(app_settings.models_cache_dir)
# Ensure cache directory exists
app_settings.models_cache_dir.mkdir(exist_ok=True, parents=True)
logger.info(f"[RERANKER] Loading Cross Encoder model: {self.model_name}")
self._model = CrossEncoder(
self.model_name,
cache_folder=cache_path,
token=settings.hf_token,
)
logger.info("[RERANKER] Cross Encoder model loaded successfully")
return self._model
@property
def name(self) -> str:
return f"CrossEncoderReranker({self.model_name})"
async def rerank(
self,
query: str,
results: list[SearchResult],
top_k: int,
) -> list[SearchResult]:
"""Rerank results using local Cross Encoder model.
Args:
query: The search query.
results: Initial search results.
top_k: Number of results to return.
Returns:
Reranked results sorted by relevance score.
"""
if not results:
return []
print(
f"[RERANKER] Cross Encoder reranking {len(results)} documents, "
f"query: '{query[:50]}...', top_k: {top_k}"
)
start_time = time.time()
try:
# Prepare query-document pairs for scoring
# CrossEncoder expects list of [query, document] pairs
pairs = [[query, result.content] for result in results]
# Get relevance scores (higher = more relevant)
scores = self.model.predict(pairs)
elapsed = time.time() - start_time
logger.info(f"[RERANKER] Cross Encoder reranking completed in {elapsed:.3f}s")
# Create new results with cross-encoder scores
scored_results = []
for i, (result, score) in enumerate(zip(results, scores)):
logger.debug(
f"[RERANKER] CrossEncoder score for doc {i}: {score:.4f} "
f"(original: {result.score:.4f}) - '{result.content[:30]}...'"
)
scored_results.append(
SearchResult(
content=result.content,
score=float(score), # Use cross-encoder score
metadata=result.metadata,
parent_doc_id=result.parent_doc_id,
)
)
# Sort by cross-encoder score (descending)
scored_results.sort(key=lambda x: x.score, reverse=True)
# Log top results
for i, r in enumerate(scored_results[:3]):
logger.debug(
f"[RERANKER] Rank #{i + 1}: score={r.score:.4f}, content='{r.content[:50]}...'"
)
return scored_results[:top_k]
except Exception as e:
logger.error(f"[RERANKER] Cross Encoder reranking failed: {e!s}")
return results[:top_k]
def warmup(self) -> None:
"""Trigger model download and loading."""
logger.info(f"[RERANKER] Cross Encoder warmup: loading model {self.model_name}")
_ = self.model
logger.info(f"[RERANKER] Cross Encoder ready: {self.model_name}")
class RerankService:
"""Service for managing reranking operations.
Orchestrates reranking using a configured reranker provider.
Supports both Cohere API and local Cross Encoder models.
"""
def __init__(self, settings: RAGSettings):
"""Initialize the rerank service.
Args:
settings: RAG configuration settings containing reranker config.
"""
self.settings = settings
config = settings.reranker_config # type: ignore[attr-defined]
self._reranker: BaseReranker | None = None
if config.model == "cross_encoder":
self._reranker = CrossEncoderReranker()
logger.info("[RERANKER] Using Cross Encoder reranker")
if self._reranker is None:
logger.warning(
f"[RERANKER] No reranker configured (model: {config.model}). "
"Reranking will be skipped."
)
@property
def reranker(self) -> BaseReranker | None:
"""Return the configured reranker, if any."""
return self._reranker
@property
def is_enabled(self) -> bool:
"""Check if reranking is enabled."""
return self._reranker is not None
async def rerank(
self,
query: str,
results: list[SearchResult],
top_k: int,
) -> list[SearchResult]:
"""Rerank search results if a reranker is configured.
Args:
query: The search query.
results: Initial search results to rerank.
top_k: Number of results to return.
Returns:
Reranked results if reranker is configured, otherwise original results.
"""
if not self._reranker:
logger.debug("[RERANKER] No reranker configured, returning original results")
return results[:top_k]
print(
f"[RERANKER] Starting reranking with {self._reranker.name}, "
f"query: '{query[:50]}...', results: {len(results)}, top_k: {top_k}"
)
# Log pre-reranking scores
for i, r in enumerate(results[:5]):
logger.debug(
f"[RERANKER] Pre-rerank #{i + 1}: score={r.score:.4f}, "
f"content='{r.content[:50]}...'"
)
reranked = await self._reranker.rerank(query, results, top_k)
# Log post-reranking scores
for i, r in enumerate(reranked[:5]):
logger.debug(
f"[RERANKER] Post-rerank #{i + 1}: score={r.score:.4f}, "
f"content='{r.content[:50]}...'"
)
return reranked
def warmup(self) -> None:
"""Initialize the reranker model if configured."""
if self._reranker:
logger.info(f"[RERANKER] Warming up {self._reranker.name}")
self._reranker.warmup()
logger.info(f"[RERANKER] {self._reranker.name} warmup complete")
+381
View File
@@ -0,0 +1,381 @@
from __future__ import annotations
import hashlib
import logging
import time
from abc import ABC, abstractmethod
from backend.services.rag.config import RAGSettings
from backend.services.rag.models import SearchResult
from backend.services.rag.reranker import RerankService
from backend.services.rag.vectorstore import BaseVectorStore
logger = logging.getLogger(__name__)
class BaseRetrievalService(ABC):
"""Abstract base class for retrieval service implementations.
Defines the interface for querying the vector store and retrieving
relevant document chunks based on a query.
"""
@abstractmethod
async def retrieve(
self,
query: str,
collection_name: str,
limit: int = 5,
min_score: float = 0.0,
filter: str = "",
) -> list[SearchResult]:
"""Execute the retrieval pipeline to find relevant chunks.
Args:
query: The search query text.
collection_name: Name of the collection to search in.
limit: Maximum number of results to return.
min_score: Minimum similarity score threshold (0.0 to 1.0).
filter: Optional filter expression for the search.
Returns:
List of SearchResult objects sorted by relevance.
"""
pass
@abstractmethod
async def retrieve_by_document(
self, query: str, collection_name: str, document_id: str, limit: int = 3
) -> list[SearchResult]:
"""Specialized retrieval restricted to a single document.
Useful for "Chat with this PDF" functionality where results
should only come from a specific document.
Args:
query: The search query text.
collection_name: Name of the collection to search in.
document_id: ID of the document to restrict search to.
limit: Maximum number of results to return.
Returns:
List of SearchResult objects from the specified document.
"""
pass
class RetrievalService(BaseRetrievalService):
"""High-level retrieval service with multi-stage pipeline.
Handles query execution against any vector store backend, including
vector search, hybrid BM25 fusion, score filtering, and reranking.
"""
def __init__(
self,
vector_store: BaseVectorStore,
settings: RAGSettings,
rerank_service: RerankService | None = None,
):
"""Initialize the Milvus retrieval service.
Args:
vector_store: The vector store to query.
settings: RAG configuration settings.
rerank_service: Optional reranking service for improved results.
"""
self.store = vector_store
self.settings = settings
self.rerank_service = rerank_service
self._reranker_enabled = rerank_service is not None and rerank_service.is_enabled
self._hybrid_enabled = settings.enable_hybrid_search
self._bm25_index: dict[str, object] = {} # collection_name -> BM25 index
@staticmethod
def _rrf_fuse(
vector_results: list[SearchResult],
bm25_results: list[SearchResult],
k: int = 60,
) -> list[SearchResult]:
"""Reciprocal Rank Fusion of vector and BM25 results."""
scores: dict[str, float] = {}
result_map: dict[str, SearchResult] = {}
for rank, r in enumerate(vector_results):
key = (
f"{r.parent_doc_id}:{r.metadata.get('chunk_num', '')}"
if r.parent_doc_id
else hashlib.md5(r.content.encode()).hexdigest()
)
scores[key] = scores.get(key, 0) + 1.0 / (k + rank + 1)
result_map[key] = r
for rank, r in enumerate(bm25_results):
key = (
f"{r.parent_doc_id}:{r.metadata.get('chunk_num', '')}"
if r.parent_doc_id
else hashlib.md5(r.content.encode()).hexdigest()
)
scores[key] = scores.get(key, 0) + 1.0 / (k + rank + 1)
if key not in result_map:
result_map[key] = r
sorted_keys = sorted(scores, key=lambda x: scores[x], reverse=True)
return [
SearchResult(
content=result_map[key].content,
score=scores[key],
metadata=result_map[key].metadata,
parent_doc_id=result_map[key].parent_doc_id,
)
for key in sorted_keys
]
async def _bm25_search(
self, query: str, collection_name: str, limit: int
) -> list[SearchResult]:
"""BM25 keyword search over stored documents."""
try:
from rank_bm25 import BM25Okapi
except ImportError:
logger.warning("rank-bm25 not installed, skipping BM25 search")
return []
# Get all documents for BM25 (cached per collection)
docs = await self.store.get_documents(collection_name)
if not docs:
return []
# Build corpus from stored content via vector store search with high limit
all_results = await self.store.search(
collection_name=collection_name, query=query, limit=min(limit * 10, 100)
)
if not all_results:
return []
corpus = [r.content.lower().split() for r in all_results]
bm25 = BM25Okapi(corpus)
query_tokens = query.lower().split()
bm25_scores = bm25.get_scores(query_tokens)
scored = sorted(zip(all_results, bm25_scores), key=lambda x: x[1], reverse=True)
return [
SearchResult(
content=r.content,
score=float(s),
metadata=r.metadata,
parent_doc_id=r.parent_doc_id,
)
for r, s in scored[:limit]
if s > 0
]
async def retrieve(
self,
query: str,
collection_name: str,
limit: int = 5,
min_score: float = 0.0,
filter: str = "",
use_reranker: bool = False,
) -> list[SearchResult]:
"""Execute the retrieval pipeline: Vector Search + Reranking (optional) + Filtering.
Args:
query: The search query text.
collection_name: Name of the collection to search in.
limit: Maximum number of results to return.
min_score: Minimum similarity score threshold (0.0 to 1.0).
filter: Optional filter expression for the search.
use_reranker: Whether to use reranking (if configured).
Returns:
List of SearchResult objects sorted by relevance.
"""
# Determine if we should actually use reranking
should_rerank = use_reranker and self._reranker_enabled
# Fetch more results if reranking is enabled (reranker will reduce)
# We fetch 3x results to give reranker room to pick best ones
fetch_multiplier = 3 if should_rerank else 2
logger.info(
f"[RETRIEVAL] Query: '{query[:50]}...', collection: {collection_name}, "
f"limit: {limit}, filter: '{filter}', rerank: {should_rerank}"
)
start_time = time.time()
# Step 1: Execute Vector Search via the Vector Store
raw_results = await self.store.search(
collection_name=collection_name,
query=query,
filter=filter,
limit=limit * fetch_multiplier,
)
search_time = time.time() - start_time
logger.info(
f"[RETRIEVAL] Vector search completed in {search_time:.3f}s, "
f"found {len(raw_results)} results"
)
# Step 1b: Hybrid search (BM25 + vector fusion) if enabled
if self._hybrid_enabled:
bm25_results = await self._bm25_search(query, collection_name, limit * fetch_multiplier)
if bm25_results:
raw_results = self._rrf_fuse(raw_results, bm25_results)
logger.info(f"[RETRIEVAL] Hybrid search: fused {len(raw_results)} results")
# Log initial results
for i, r in enumerate(raw_results[:3]):
logger.debug(
f"[RETRIEVAL] Initial result #{i + 1}: score={r.score:.4f}, "
f"content='{r.content[:50]}...'"
)
results = raw_results
# Step 2: Apply reranking if enabled and requested
if should_rerank and self.rerank_service:
logger.info("[RETRIEVAL] Applying reranking...")
rerank_start = time.time()
# Rerank the results - fetches more initially so reranker can pick best
results = await self.rerank_service.rerank(
query=query,
results=raw_results,
top_k=limit * 2, # Get more from reranker before filtering
)
rerank_time = time.time() - rerank_start
logger.info(
f"[RETRIEVAL] Reranking completed in {rerank_time:.3f}s, "
f"returned {len(results)} results"
)
elif use_reranker and not self._reranker_enabled:
logger.warning("[RETRIEVAL] Reranking requested but not configured - skipping")
# Step 3: Post-processing: Filter by score
# Cosine similarity is higher = better.
filtered_results = [res for res in results if res.score >= min_score]
# Step 4: Deduplicate — keep highest-scored result per unique chunk
seen_keys: set[str] = set()
deduped_results: list[SearchResult] = []
for r in filtered_results:
key = (
f"{r.parent_doc_id}:{r.metadata.get('chunk_num', '')}"
if r.parent_doc_id
else hashlib.md5(r.content.encode()).hexdigest()
)
if key not in seen_keys:
seen_keys.add(key)
deduped_results.append(r)
if len(deduped_results) < len(filtered_results):
logger.info(
f"[RETRIEVAL] Deduplicated: {len(filtered_results)} -> {len(deduped_results)} results"
)
# Log final results
for i, r in enumerate(deduped_results[:3]):
logger.debug(
f"[RETRIEVAL] Final result #{i + 1}: score={r.score:.4f}, "
f"content='{r.content[:50]}...'"
)
# Apply final limit
final_results = deduped_results[:limit]
total_time = time.time() - start_time
logger.info(
f"[RETRIEVAL] Total retrieval time: {total_time:.3f}s, "
f"returning {len(final_results)} results"
)
return final_results
async def retrieve_multi(
self,
query: str,
collection_names: list[str],
limit: int = 5,
min_score: float = 0.0,
use_reranker: bool = False,
) -> list[SearchResult]:
"""Search across multiple collections and merge results.
Searches each collection, merges results, sorts by score.
"""
all_results: list[SearchResult] = []
for name in collection_names:
try:
results = await self.retrieve(
query=query,
collection_name=name,
limit=limit,
min_score=min_score,
use_reranker=use_reranker,
)
# Tag results with collection name in metadata
for r in results:
r.metadata["collection"] = name
all_results.extend(results)
except Exception as e:
logger.warning(f"[RETRIEVAL] Failed to search collection '{name}': {e}")
all_results.sort(key=lambda r: r.score, reverse=True)
# Deduplicate across collections
seen_keys: set[str] = set()
deduped: list[SearchResult] = []
for r in all_results:
key = (
f"{r.parent_doc_id}:{r.metadata.get('chunk_num', '')}"
if r.parent_doc_id
else hashlib.md5(r.content.encode()).hexdigest()
)
if key not in seen_keys:
seen_keys.add(key)
deduped.append(r)
return deduped[:limit]
async def retrieve_by_document(
self,
query: str,
collection_name: str,
document_id: str,
limit: int = 3,
use_reranker: bool = False,
) -> list[SearchResult]:
"""Specialized retrieval restricted to a single document.
Useful for "Chat with this PDF" functionality where results
should only come from a specific document.
Args:
query: The search query text.
collection_name: Name of the collection to search in.
document_id: ID of the document to restrict search to.
limit: Maximum number of results to return.
use_reranker: Whether to use reranking (if configured).
Returns:
List of SearchResult objects from the specified document.
"""
# Sanitize document_id to prevent filter injection
sanitized_id = document_id.replace('"', "").replace("\\", "")
filter_expr = f'parent_doc_id == "{sanitized_id}"'
logger.info(
f"[RETRIEVAL] Retrieve by document: doc_id={document_id}, "
f"query='{query[:30]}...', limit={limit}, rerank={use_reranker}"
)
return await self.retrieve(
query=query,
collection_name=collection_name,
limit=limit,
filter=filter_expr,
use_reranker=use_reranker,
)
+246
View File
@@ -0,0 +1,246 @@
import logging
import re
from abc import ABC, abstractmethod
from typing import Any
from backend.schemas.rag import RAGDocumentItem, RAGDocumentList
from backend.services.rag.models import (
CollectionInfo,
Document,
DocumentInfo,
DocumentPageChunk,
SearchResult,
)
logger = logging.getLogger(__name__)
_COLLECTION_NAME_RE = re.compile(r"^[a-zA-Z][a-zA-Z0-9_]{0,63}$")
_RESERVED_COLLECTION_NAMES = frozenset({"all"})
class BaseVectorStore(ABC):
"""Abstract base class for vector store implementations."""
@abstractmethod
async def insert_document(self, collection_name: str, document: Document) -> None:
"""Embeds and stores document chunks."""
@abstractmethod
async def search(
self, collection_name: str, query: str, limit: int = 4, filter: str = ""
) -> list[SearchResult]:
"""Retrieves similar chunks based on a text query."""
@abstractmethod
async def delete_collection(self, collection_name: str) -> None:
"""Removes a collection and all its data."""
@abstractmethod
async def delete_document(self, collection_name: str, document_id: str) -> None:
"""Removes all chunks associated with a document ID."""
@abstractmethod
async def get_collection_info(self, collection_name: str) -> CollectionInfo:
"""Returns metadata and stats about a collection."""
@abstractmethod
async def list_collections(self) -> list[str]:
"""Returns list of all collection names."""
@abstractmethod
async def get_documents(self, collection_name: str) -> list[DocumentInfo]:
"""Returns list of unique documents in a collection."""
async def get_document_list(self, collection_name: str) -> RAGDocumentList:
"""Returns documents as API-ready list response."""
docs = await self.get_documents(collection_name)
return RAGDocumentList(
items=[
RAGDocumentItem(
document_id=doc.document_id,
filename=doc.filename,
filesize=doc.filesize,
filetype=doc.filetype,
chunk_count=doc.chunk_count,
additional_info=doc.additional_info,
)
for doc in docs
],
total=len(docs),
)
async def create_collection(self, name: str) -> None:
"""Validate the name and create the collection.
Raises:
ValueError: If name is invalid or reserved.
"""
if not _COLLECTION_NAME_RE.match(name):
raise ValueError(
"Collection name must start with a letter and contain only "
"letters, numbers, and underscores (max 64 chars)"
)
if name.lower() in _RESERVED_COLLECTION_NAMES:
raise ValueError(f"'{name}' is a reserved collection name")
await self._ensure_collection(name)
def _build_chunk_metadata(
self, chunk: "DocumentPageChunk", document: Document
) -> dict[str, Any]:
"""Build metadata dict for a chunk."""
meta = {
"page_num": chunk.page_num,
"chunk_num": chunk.chunk_num,
"has_images": bool(getattr(chunk, "images", None)),
"image_count": len(getattr(chunk, "images", [])),
**document.metadata.model_dump(),
}
return meta
def _sanitize_id(self, document_id: str) -> str:
"""Sanitize document_id to prevent filter injection."""
return document_id.replace('"', "").replace("\\", "")
def _group_documents(self, results: list[dict[str, Any]]) -> list[DocumentInfo]:
"""Group query results by parent_doc_id into DocumentInfo list."""
doc_map: dict[str, dict[str, Any]] = {}
for item in results:
doc_id = item.get("parent_doc_id")
metadata = item.get("metadata", {})
if doc_id and doc_id not in doc_map:
doc_map[doc_id] = {
"document_id": doc_id,
"filename": metadata.get("filename"),
"filesize": metadata.get("filesize"),
"filetype": metadata.get("filetype"),
"additional_info": {
"source_path": metadata.get("source_path", ""),
"content_hash": metadata.get("content_hash", ""),
**(metadata.get("additional_info") or {}),
},
"chunk_count": 0,
}
if doc_id:
doc_map[doc_id]["chunk_count"] += 1
return [
DocumentInfo(
document_id=d["document_id"],
filename=d.get("filename"),
filesize=d.get("filesize"),
filetype=d.get("filetype"),
chunk_count=d["chunk_count"],
additional_info=d.get("additional_info"),
)
for d in doc_map.values()
]
from pymilvus import AsyncMilvusClient, DataType
from backend.core.config import settings as app_settings
from backend.services.rag.config import RAGSettings
from backend.services.rag.embeddings import EmbeddingService
class MilvusVectorStore(BaseVectorStore):
"""Milvus vector store implementation."""
def __init__(self, settings: RAGSettings, embedding_service: EmbeddingService):
self.settings = settings
self.embedder = embedding_service
self.client = AsyncMilvusClient(
uri=app_settings.milvus_uri, token=app_settings.milvus_token
)
async def _ensure_collection(self, name: str) -> None:
if not await self.client.has_collection(name):
schema = self.client.create_schema(auto_id=False)
schema.add_field("id", DataType.VARCHAR, is_primary=True, max_length=100)
schema.add_field("parent_doc_id", DataType.VARCHAR, max_length=100)
schema.add_field("content", DataType.VARCHAR, max_length=65535)
schema.add_field(
"vector", DataType.FLOAT_VECTOR, dim=self.settings.embeddings_config.dim
)
schema.add_field("metadata", DataType.JSON)
await self.client.create_collection(
name, schema=schema, metric_type="COSINE"
)
indexes = await self.client.list_indexes(name)
if not indexes:
index_params = self.client.prepare_index_params()
index_params.add_index(
field_name="vector", index_type="AUTOINDEX", metric_type="COSINE"
)
await self.client.create_index(
collection_name=name, index_params=index_params
)
await self.client.load_collection(name)
async def insert_document(self, collection_name: str, document: Document) -> None:
await self._ensure_collection(collection_name)
if not document.chunked_pages:
raise ValueError("Document has no chunked pages.")
vectors = self.embedder.embed_document(document)
data = [
{
"id": chunk.chunk_id,
"parent_doc_id": chunk.parent_doc_id,
"content": chunk.chunk_content,
"vector": vectors[i],
"metadata": self._build_chunk_metadata(chunk, document),
}
for i, chunk in enumerate(document.chunked_pages)
]
await self.client.insert(collection_name, data=data)
async def search(
self, collection_name: str, query: str, limit: int = 4, filter: str = ""
) -> list[SearchResult]:
query_vector = self.embedder.embed_query(query)
results = await self.client.search(
collection_name=collection_name,
data=[query_vector],
limit=limit,
filter=filter,
output_fields=["content", "parent_doc_id", "metadata"],
)
return [
SearchResult(
content=hit["entity"]["content"],
score=hit["distance"],
metadata=hit["entity"]["metadata"],
parent_doc_id=hit["entity"]["parent_doc_id"],
)
for hit in results[0]
]
async def get_collection_info(self, collection_name: str) -> CollectionInfo:
count = await self.client.get_collection_stats(collection_name)
return CollectionInfo(
name=collection_name,
total_vectors=count.get("row_count", 0),
dim=self.settings.embeddings_config.dim,
)
async def delete_collection(self, collection_name: str) -> None:
await self.client.drop_collection(collection_name)
async def delete_document(self, collection_name: str, document_id: str) -> None:
sanitized = self._sanitize_id(document_id)
await self.client.delete(
collection_name=collection_name, filter=f'parent_doc_id == "{sanitized}"'
)
async def get_documents(self, collection_name: str) -> list[DocumentInfo]:
await self._ensure_collection(collection_name)
results = await self.client.query(
collection_name=collection_name,
filter="",
output_fields=["parent_doc_id", "metadata"],
limit=10000,
)
return self._group_documents(results)
async def list_collections(self) -> list[str]:
result: list[str] = await self.client.list_collections()
return result
+280
View File
@@ -0,0 +1,280 @@
"""RAG document service (PostgreSQL async).
Contains business logic for tracking RAG document ingestion, status updates,
file downloads, cascading deletions across DB+vector store+file storage, and
upload-and-dispatch orchestration for the Celery/Taskiq/ARQ worker.
"""
import logging
import os
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from uuid import UUID
from sqlalchemy.ext.asyncio import AsyncSession
from backend.core.config import settings
from backend.core.exceptions import BadRequestError, NotFoundError
from backend.db.models.rag_document import RAGDocument
from backend.repositories import rag_document_repo
from backend.schemas.rag import RAGIngestResponse, RAGTrackedDocumentItem, RAGTrackedDocumentList
from backend.services.file_storage import get_file_storage
from backend.services.rag.config import get_supported_formats
logger = logging.getLogger(__name__)
class RAGDocumentService:
"""Service for RAG document tracking and lifecycle management."""
def __init__(self, db: AsyncSession):
self.db = db
async def list_documents(
self,
collection_name: str | None = None,
) -> RAGTrackedDocumentList:
"""List tracked RAG documents, optionally filtered by collection."""
docs = await rag_document_repo.get_all(self.db, collection_name)
return RAGTrackedDocumentList(
items=[
RAGTrackedDocumentItem(
id=str(d.id),
collection_name=d.collection_name,
filename=d.filename,
filesize=d.filesize,
filetype=d.filetype,
status=d.status,
error_message=d.error_message,
vector_document_id=d.vector_document_id,
chunk_count=d.chunk_count,
has_file=bool(d.storage_path),
created_at=d.created_at.isoformat() if d.created_at else None,
completed_at=d.completed_at.isoformat() if d.completed_at else None,
)
for d in docs
],
total=len(docs),
)
async def get_document(self, doc_id: str) -> RAGDocument:
"""Get a RAG document by ID.
Raises:
NotFoundError: If document does not exist.
"""
doc = await rag_document_repo.get_by_id(self.db, UUID(doc_id))
if not doc:
raise NotFoundError(
message="Document not found",
details={"doc_id": doc_id},
)
return doc
async def create_document(
self,
*,
collection_name: str,
filename: str,
filesize: int,
filetype: str,
storage_path: str | None = None,
) -> RAGDocument:
"""Create a new RAG document tracking record."""
return await rag_document_repo.create(
self.db,
collection_name=collection_name,
filename=filename,
filesize=filesize,
filetype=filetype,
storage_path=storage_path or "",
)
async def dispatch_upload(
self,
*,
collection_name: str,
file_data: bytes,
filename: str,
replace: bool,
vector_store: Any,
) -> RAGIngestResponse:
"""Validate, persist, and queue an uploaded file for ingestion.
Performs:
1. file-extension and size validation (against ``settings``);
2. permanent storage via ``FileStorage``;
3. RAGDocument tracking-record creation;
4. lazy creation of the target vector collection;
5. tmp-copy under ``MEDIA_DIR/_rag_tmp`` (shared with worker container);
6. dispatch of the ingestion task on the configured task backend.
"""
allowed = get_supported_formats(getattr(settings, "pdf_parser", "pymupdf"))
max_size = settings.max_upload_size_mb * 1024 * 1024
ext = Path(filename).suffix.lower()
if ext not in allowed:
raise BadRequestError(
message=f"File type '{ext}' not supported",
details={"ext": ext, "allowed": sorted(allowed)},
)
if len(file_data) > max_size:
raise BadRequestError(
message=f"File too large. Maximum {settings.max_upload_size_mb}MB.",
details={"size": len(file_data), "max_mb": settings.max_upload_size_mb},
)
storage = get_file_storage()
storage_path = await storage.save(f"rag/{collection_name}", filename, file_data)
rag_doc = await self.create_document(
collection_name=collection_name,
filename=filename,
filesize=len(file_data),
filetype=ext.lstrip("."),
storage_path=storage_path,
)
doc_id = rag_doc.id
# Ensure the target collection exists before the worker starts
await vector_store.create_collection(collection_name)
# Stage the upload in the volume shared with the worker container
tmp_dir = os.path.join(str(settings.media_dir), "_rag_tmp")
os.makedirs(tmp_dir, exist_ok=True)
tmp_path = os.path.join(tmp_dir, f"{doc_id!s}{ext}")
with open(tmp_path, "wb") as f:
f.write(file_data)
from backend.worker.dispatcher import get_dispatcher
get_dispatcher().delay(
"ingest_document_task",
rag_document_id=str(doc_id),
collection_name=collection_name,
filepath=tmp_path,
source_path=filename,
replace=replace,
)
return RAGIngestResponse(
id=str(doc_id),
status="processing",
filename=filename,
collection=collection_name,
message="File accepted. Processing in background.",
)
async def complete_ingestion(
self,
doc_id: str,
vector_document_id: str,
chunk_count: int = 0,
) -> None:
"""Mark a document as successfully ingested."""
doc = await self.get_document(doc_id)
await rag_document_repo.update_status(
self.db,
doc.id,
status="done",
vector_document_id=vector_document_id,
chunk_count=chunk_count,
completed_at=datetime.now(UTC),
)
async def fail_ingestion(self, doc_id: str, error_message: str) -> None:
"""Mark a document ingestion as failed."""
doc = await self.get_document(doc_id)
await rag_document_repo.update_status(
self.db,
doc.id,
status="error",
error_message=error_message,
completed_at=datetime.now(UTC),
)
async def retry_ingestion(self, doc_id: str) -> RAGDocument:
"""Reset a failed document for re-ingestion.
Raises:
NotFoundError: If document does not exist.
ValueError: If document status is not 'error'.
"""
doc = await self.get_document(doc_id)
if doc.status != "error":
raise ValueError("Only failed documents can be retried")
updated = await rag_document_repo.update_status(
self.db,
doc.id,
status="processing",
error_message="",
completed_at=None,
)
if updated is None:
raise NotFoundError(message="Document not found", details={"doc_id": doc_id})
return updated
async def delete_document(
self,
doc_id: str,
ingestion_service: Any = None,
) -> None:
"""Delete a document with cascading cleanup.
Removes the record from the database and attempts to clean up
the vector store entry and stored file. Failures in cleanup
are logged but do not prevent the DB deletion.
"""
doc = await self.get_document(doc_id)
# Cascade: vector store
if doc.vector_document_id and ingestion_service:
try:
await ingestion_service.remove_document(doc.collection_name, doc.vector_document_id)
except Exception as e:
logger.warning(f"Failed to delete from vector store: {e}")
# Cascade: file storage
if doc.storage_path:
try:
storage = get_file_storage()
await storage.delete(doc.storage_path)
except Exception as e:
logger.warning(f"Failed to delete file: {e}")
# Cascade: DB record
await rag_document_repo.delete(self.db, doc.id)
async def delete_by_collection(self, collection_name: str) -> int:
"""Delete all RAG document records for a collection.
Returns:
Number of deleted records.
"""
return await rag_document_repo.delete_by_collection(self.db, collection_name)
async def get_download_info(self, doc_id: str) -> tuple[str, str, str]:
"""Get file download information for a document.
Returns:
Tuple of (file_path, filename, mime_type).
Raises:
NotFoundError: If document or its file does not exist.
"""
doc = await self.get_document(doc_id)
if not doc.storage_path:
raise NotFoundError(message="No file stored for this document")
storage = get_file_storage()
file_path = storage.get_full_path(doc.storage_path)
if not file_path:
raise NotFoundError(message="File not found on disk")
mime_map = {
"pdf": "application/pdf",
"docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
"txt": "text/plain",
"md": "text/markdown",
}
mime_type = mime_map.get(doc.filetype, "application/octet-stream")
return str(file_path), doc.filename, mime_type
+49
View File
@@ -0,0 +1,49 @@
"""RAG status streaming — Redis pub/sub fan-out for SSE clients."""
import asyncio
import logging
from collections.abc import AsyncIterator
import redis.asyncio as aioredis
from fastapi.sse import ServerSentEvent
from backend.core.config import settings
logger = logging.getLogger(__name__)
class RAGStatusService:
"""Streams RAG ingestion status events from Redis pub/sub."""
CHANNEL = "rag_status"
async def stream_events(self) -> AsyncIterator[ServerSentEvent]:
"""Yield ``ServerSentEvent`` items as they arrive on the ``rag_status`` channel.
The Redis client is created per-stream (one connection per subscriber). Cleanup is
guaranteed via ``finally`` even if the consumer disconnects mid-stream.
"""
client = aioredis.from_url(settings.valkey_url) # type: ignore[no-untyped-call]
pubsub = client.pubsub()
await pubsub.subscribe(self.CHANNEL)
event_id = 0
try:
async for message in pubsub.listen():
if message["type"] != "message":
continue
payload = message["data"]
data = payload.decode() if isinstance(payload, bytes) else payload
event_id += 1
yield ServerSentEvent(raw_data=data, event="status", id=str(event_id))
except asyncio.CancelledError:
# Client disconnected — propagate cancellation cleanly
raise
except Exception as exc:
logger.warning("RAG SSE stream error: %s", exc)
finally:
try:
await pubsub.unsubscribe(self.CHANNEL)
await client.aclose()
except Exception as exc:
logger.debug("RAG SSE cleanup error: %s", exc)
+152
View File
@@ -0,0 +1,152 @@
"""RAG sync service (PostgreSQL async).
Contains business logic for managing RAG synchronization operations
and their associated log entries.
"""
from datetime import UTC, datetime
from uuid import UUID
from sqlalchemy.ext.asyncio import AsyncSession
from backend.core.exceptions import NotFoundError
from backend.db.models.sync_log import SyncLog
from backend.repositories import sync_log_repo
from backend.schemas.rag import RAGSyncLogItem, RAGSyncLogList
class RAGSyncService:
"""Service for RAG sync operation management."""
def __init__(self, db: AsyncSession):
self.db = db
async def list_sync_logs(
self,
collection_name: str | None = None,
limit: int = 20,
) -> RAGSyncLogList:
"""List sync operation logs, optionally filtered by collection."""
logs = await sync_log_repo.get_all(self.db, collection_name=collection_name, limit=limit)
return RAGSyncLogList(
items=[
RAGSyncLogItem(
id=str(log.id),
source=log.source,
collection_name=log.collection_name,
status=log.status,
mode=log.mode,
total_files=log.total_files,
ingested=log.ingested,
updated=log.updated,
skipped=log.skipped,
failed=log.failed,
error_message=log.error_message,
started_at=log.started_at.isoformat() if log.started_at else None,
completed_at=log.completed_at.isoformat() if log.completed_at else None,
)
for log in logs
],
total=len(logs),
)
async def get_sync_log(self, sync_id: str) -> SyncLog:
"""Get a sync log by ID.
Raises:
NotFoundError: If sync log does not exist.
"""
log = await sync_log_repo.get_by_id(self.db, UUID(sync_id))
if not log:
raise NotFoundError(
message="Sync log not found",
details={"sync_id": sync_id},
)
return log
async def create_sync_log(
self,
*,
source: str,
collection_name: str,
mode: str,
) -> SyncLog:
"""Create a new sync log entry."""
return await sync_log_repo.create(
self.db,
source=source,
collection_name=collection_name,
mode=mode,
)
async def start_local_sync(
self,
*,
collection_name: str,
mode: str,
path: str | None,
) -> SyncLog:
"""Persist a sync log and dispatch the local-sync task on the configured backend."""
sync_log = await self.create_sync_log(
source="local",
collection_name=collection_name,
mode=mode,
)
from backend.worker.dispatcher import get_dispatcher
get_dispatcher().delay(
"sync_collection_task",
sync_log_id=str(sync_log.id),
source="local",
collection_name=collection_name,
mode=mode,
path=path,
)
return sync_log
async def complete_sync(
self,
sync_id: str,
*,
status: str,
total_files: int = 0,
ingested: int = 0,
updated: int = 0,
skipped: int = 0,
failed: int = 0,
error_message: str | None = None,
) -> None:
"""Mark a sync operation as completed (done or error)."""
log = await self.get_sync_log(sync_id)
await sync_log_repo.update_status(
self.db,
log.id,
status=status,
total_files=total_files,
ingested=ingested,
updated=updated,
skipped=skipped,
failed=failed,
error_message=error_message,
completed_at=datetime.now(UTC),
)
async def cancel_sync(self, sync_id: str) -> SyncLog:
"""Cancel a running sync operation.
Raises:
NotFoundError: If sync log does not exist.
ValueError: If sync is not in 'running' state.
"""
log = await self.get_sync_log(sync_id)
if log.status != "running":
raise ValueError("Sync is not running")
cancelled = await sync_log_repo.update_status(
self.db,
log.id,
status="cancelled",
completed_at=datetime.now(UTC),
)
if cancelled is None:
raise NotFoundError(message="Sync log not found", details={"sync_id": sync_id})
return cancelled
+189
View File
@@ -0,0 +1,189 @@
"""Sync source service (PostgreSQL async).
Contains business logic for managing RAG sync source configurations
and triggering sync operations.
"""
import json
from datetime import UTC, datetime
from uuid import UUID
from sqlalchemy.ext.asyncio import AsyncSession
from backend.core.exceptions import BadRequestError, NotFoundError
from backend.db.models.sync_log import SyncLog
from backend.db.models.sync_source import SyncSource
from backend.repositories import sync_log_repo
from backend.repositories import sync_source_repo
from backend.schemas.sync_source import (
ConnectorConfigField,
ConnectorInfo,
ConnectorList,
SyncSourceCreate,
SyncSourceList,
SyncSourceRead,
SyncSourceUpdate,
)
from backend.rag.connectors import CONNECTOR_REGISTRY
class SyncSourceService:
"""Service for managing sync source configurations."""
def __init__(self, db: AsyncSession):
self.db = db
def _to_read(self, s: SyncSource) -> SyncSourceRead:
return SyncSourceRead(
id=str(s.id),
name=s.name,
connector_type=s.connector_type,
collection_name=s.collection_name,
config=s.config
if isinstance(s.config, dict)
else json.loads(s.config)
if s.config
else {},
sync_mode=s.sync_mode,
schedule_minutes=s.schedule_minutes,
is_active=s.is_active,
last_sync_at=s.last_sync_at.isoformat() if s.last_sync_at else None,
last_sync_status=s.last_sync_status,
last_error=s.last_error,
created_at=s.created_at.isoformat() if s.created_at else None,
)
async def list_sources(
self,
is_active: bool | None = None,
) -> SyncSourceList:
"""List all sync sources, optionally filtered by active status."""
sources = await sync_source_repo.get_all(self.db, is_active=is_active)
return SyncSourceList(items=[self._to_read(s) for s in sources], total=len(sources))
async def get_source(self, source_id: str) -> SyncSource:
"""Get a sync source by ID.
Raises:
NotFoundError: If sync source does not exist.
"""
source = await sync_source_repo.get_by_id(self.db, UUID(source_id))
if not source:
raise NotFoundError(
message="Sync source not found",
details={"source_id": source_id},
)
return source
async def create_source(self, data: SyncSourceCreate) -> SyncSourceRead:
"""Create a new sync source.
Validates the connector type and its configuration before creating.
Raises:
BadRequestError: If connector type is unknown or config is invalid.
"""
if data.connector_type not in CONNECTOR_REGISTRY:
raise BadRequestError(
message=f"Unknown connector type: {data.connector_type}",
details={"connector_type": data.connector_type},
)
connector_cls = CONNECTOR_REGISTRY[data.connector_type]
connector = connector_cls()
is_valid, error = await connector.validate_config(data.config)
if not is_valid:
raise BadRequestError(
message=f"Invalid connector config: {error}",
details={"connector_type": data.connector_type},
)
source = await sync_source_repo.create(
self.db,
name=data.name,
connector_type=data.connector_type,
collection_name=data.collection_name,
config=data.config,
sync_mode=data.sync_mode,
schedule_minutes=data.schedule_minutes,
)
return self._to_read(source)
async def update_source(self, source_id: str, data: SyncSourceUpdate) -> SyncSourceRead:
"""Update an existing sync source.
Raises:
NotFoundError: If sync source does not exist.
"""
await self.get_source(source_id) # verify exists
updates = data.model_dump(exclude_unset=True)
source = await sync_source_repo.update(self.db, UUID(source_id), **updates)
if source is None:
raise NotFoundError(message="Sync source not found", details={"source_id": source_id})
return self._to_read(source)
async def delete_source(self, source_id: str) -> None:
"""Delete a sync source.
Raises:
NotFoundError: If sync source does not exist.
"""
await self.get_source(source_id) # verify exists
await sync_source_repo.delete(self.db, UUID(source_id))
async def trigger_sync(self, source_id: str) -> SyncLog:
"""Trigger a manual sync — persists a SyncLog and dispatches the task.
Raises:
NotFoundError: If sync source does not exist.
"""
source = await self.get_source(source_id)
sync_log = await sync_log_repo.create(
self.db,
source=source.connector_type,
collection_name=source.collection_name,
mode=source.sync_mode,
sync_source_id=source.id,
)
from backend.worker.dispatcher import get_dispatcher
get_dispatcher().delay(
"sync_single_source_task",
source_id=source_id,
sync_log_id=str(sync_log.id),
)
return sync_log
async def update_after_sync(
self,
source_id: str,
status: str,
error: str | None = None,
) -> None:
"""Update sync source status after a sync operation completes."""
await sync_source_repo.update_sync_status(
self.db,
UUID(source_id),
last_sync_at=datetime.now(UTC),
last_sync_status=status,
last_error=error,
)
@staticmethod
def list_connectors() -> ConnectorList:
"""List available connector types with their config schemas."""
items = []
for _connector_type, connector_cls in CONNECTOR_REGISTRY.items():
schema_fields = {
field_name: ConnectorConfigField(**field_spec)
for field_name, field_spec in connector_cls.CONFIG_SCHEMA.items()
}
items.append(
ConnectorInfo(
type=connector_cls.CONNECTOR_TYPE,
name=connector_cls.DISPLAY_NAME,
config_schema=schema_fields,
enabled=True,
)
)
return ConnectorList(items=items)
View File
+26
View File
@@ -0,0 +1,26 @@
"""Task dispatcher configuration.
In the current implementation, tasks run in-process via
:class:`backend.worker.dispatcher.TaskDispatcher`. For production
deployments with multiple replicas, replace with a Redis-backed
queue (ARQ, Taskiq, or Celery).
To switch to ARQ in the future::
pip install arq
arq backend.worker.arq_settings.WorkerSettings
"""
from __future__ import annotations
from backend.core.config import settings
from backend.worker.tasks.rag_tasks import TASK_REGISTRY
# List of registered task functions for ARQ/Taskiq configuration
FUNCTIONS = list(TASK_REGISTRY.values())
# Redis connection info (for future ARQ/Taskiq worker setup)
REDIS_URL = settings.valkey_url
__all__ = ["FUNCTIONS", "REDIS_URL"]
+91
View File
@@ -0,0 +1,91 @@
"""Lightweight async task dispatcher.
Dispatches background tasks to the configured backend.
Currently uses in-process ``asyncio.create_task`` for simplicity.
A Redis-backed ARQ/Taskiq worker can replace this for production use.
"""
from __future__ import annotations
import asyncio
import logging
from typing import Any
logger = logging.getLogger(__name__)
class Task:
"""Represents a background task to be executed."""
def __init__(self, name: str, kwargs: dict[str, Any]) -> None:
self.name = name
self.kwargs = kwargs
class TaskDispatcher:
"""Dispatches background tasks.
In the current implementation, tasks are executed in-process using
``asyncio.create_task``. For production, swap the backend to use
Redis (ARQ or Taskiq) for cross-process execution.
"""
def __init__(self) -> None:
self._running: set[asyncio.Task] = set()
def delay(self, task_name: str, **kwargs: Any) -> None:
"""Enqueue a task for background execution.
This returns immediately. The task is executed asynchronously
in the event loop.
"""
task = asyncio.create_task(self._execute(task_name, kwargs))
self._running.add(task)
task.add_done_callback(self._running.discard)
logger.debug("Dispatched task: %s (kwargs=%s)", task_name, kwargs)
async def _execute(self, name: str, kwargs: dict[str, Any]) -> None:
"""Look up the task function and run it."""
from backend.worker.tasks.rag_tasks import TASK_REGISTRY
fn = TASK_REGISTRY.get(name)
if fn is None:
logger.error("Unknown task: %s", name)
return
try:
await fn({"dispatcher": self}, **kwargs)
except Exception as exc:
logger.exception("Task %s failed: %s", name, exc)
async def shutdown(self) -> None:
"""Cancel all running tasks (called on app shutdown)."""
for task in self._running:
task.cancel()
if self._running:
await asyncio.gather(*self._running, return_exceptions=True)
self._running.clear()
# Module-level singleton
_dispatcher: TaskDispatcher | None = None
def get_dispatcher() -> TaskDispatcher:
"""Return the shared TaskDispatcher singleton."""
global _dispatcher
if _dispatcher is None:
_dispatcher = TaskDispatcher()
return _dispatcher
def shutdown_dispatcher() -> None:
"""Shutdown and clear the dispatcher singleton."""
global _dispatcher
if _dispatcher is not None:
# Schedule shutdown on the running event loop
try:
loop = asyncio.get_running_loop()
loop.create_task(_dispatcher.shutdown())
except RuntimeError:
pass # No running loop — nothing to clean up
_dispatcher = None
View File
+269
View File
@@ -0,0 +1,269 @@
"""Task functions for RAG ingestion and sync operations.
Each function receives a ``ctx`` dict. Tasks are registered in
``TASK_REGISTRY`` and dispatched by :class:`TaskDispatcher`.
"""
from __future__ import annotations
import logging
from pathlib import Path
from typing import Any
from backend.core.config import settings
from backend.services.rag_document import RAGDocumentService
from backend.services.rag_sync import RAGSyncService
logger = logging.getLogger(__name__)
# Registry: maps task name -> async callable(ctx, **kwargs)
TASK_REGISTRY: dict[str, Any] = {}
def _register(fn):
"""Decorator to register a task function."""
TASK_REGISTRY[fn.__name__] = fn
return fn
@_register
async def ingest_document_task(
ctx: dict,
rag_document_id: str,
collection_name: str,
filepath: str,
source_path: str,
replace: bool = True,
) -> dict:
"""Process a single file through the RAG ingestion pipeline.
Called by :meth:`RAGDocumentService.dispatch_upload`.
Steps:
1. Build the IngestionService
2. Parse, chunk, embed, and store the document
3. Mark the tracked record as done (or error)
"""
logger.info(
"[WORKER] Starting ingestion: doc=%s, collection=%s, file=%s",
rag_document_id,
collection_name,
filepath,
)
from backend.services.rag.embeddings import EmbeddingService
from backend.services.rag.documents import DocumentProcessor
from backend.services.rag.vectorstore import MilvusVectorStore as VectorStore
from backend.services.rag.ingestion import IngestionService
try:
rag_settings = settings.rag
embed_service = EmbeddingService(settings=rag_settings)
vector_store = VectorStore(settings=rag_settings, embedding_service=embed_service)
processor = DocumentProcessor(settings=rag_settings)
ingestion = IngestionService(processor=processor, vector_store=vector_store)
result = await ingestion.ingest_file(
filepath=Path(filepath),
collection_name=collection_name,
replace=replace,
source_path=source_path,
)
# Create a DB session to update the tracked document record
from backend.core.database import open_session
async with open_session() as session:
doc_service = RAGDocumentService(db=session)
if result.status.value == "done" and result.document_id:
await doc_service.complete_ingestion(
doc_id=rag_document_id,
vector_document_id=result.document_id,
chunk_count=0, # could be tracked from result
)
elif result.status.value == "error":
await doc_service.fail_ingestion(
doc_id=rag_document_id,
error_message=result.error_message or "Unknown error",
)
logger.info(
"[WORKER] Ingestion complete: doc=%s, status=%s",
rag_document_id,
result.status,
)
return {"status": result.status.value, "document_id": result.document_id}
except Exception as exc:
logger.exception("[WORKER] Ingestion failed: doc=%s", rag_document_id)
from backend.core.database import open_session
async with open_session() as session:
doc_service = RAGDocumentService(db=session)
await doc_service.fail_ingestion(
doc_id=rag_document_id,
error_message=str(exc),
)
return {"status": "error", "message": str(exc)}
@_register
async def sync_collection_task(
ctx: dict,
sync_log_id: str,
source: str,
collection_name: str,
mode: str,
path: str | None = None,
) -> dict:
"""Sync a local directory or source into the RAG collection.
Called by :meth:`RAGSyncService.start_local_sync`.
"""
logger.info(
"[WORKER] Starting sync: log=%s, collection=%s, mode=%s, path=%s",
sync_log_id,
collection_name,
mode,
path,
)
from backend.services.rag.embeddings import EmbeddingService
from backend.services.rag.documents import DocumentProcessor
from backend.services.rag.vectorstore import MilvusVectorStore as VectorStore
from backend.services.rag.ingestion import IngestionService
try:
rag_settings = settings.rag
embed_service = EmbeddingService(settings=rag_settings)
vector_store = VectorStore(settings=rag_settings, embedding_service=embed_service)
processor = DocumentProcessor(settings=rag_settings)
ingestion = IngestionService(processor=processor, vector_store=vector_store)
ingested = 0
failed = 0
skipped = 0
if path and Path(path).is_dir():
allowed_exts = {".pdf", ".docx", ".txt", ".md"}
files = [p for p in Path(path).iterdir() if p.suffix.lower() in allowed_exts]
total = len(files)
logger.info("[WORKER] Found %d files to sync from %s", total, path)
for filepath in files:
result = await ingestion.ingest_file(
filepath=filepath,
collection_name=collection_name,
replace=(mode == "full"),
source_path=str(filepath),
)
if result.status.value == "done":
ingested += 1
elif result.status.value == "error":
failed += 1
else:
skipped += 1
else:
total = 0
logger.warning("[WORKER] No valid path provided for sync")
from backend.core.database import open_session
async with open_session() as session:
sync_service = RAGSyncService(db=session)
await sync_service.complete_sync(
sync_id=sync_log_id,
status="done" if failed == 0 else "partial",
total_files=total,
ingested=ingested,
updated=0,
skipped=skipped,
failed=failed,
)
logger.info(
"[WORKER] Sync complete: log=%s, ingested=%d, failed=%d",
sync_log_id,
ingested,
failed,
)
return {
"status": "done",
"total": total,
"ingested": ingested,
"failed": failed,
"skipped": skipped,
}
except Exception as exc:
logger.exception("[WORKER] Sync failed: log=%s", sync_log_id)
from backend.core.database import open_session
async with open_session() as session:
sync_service = RAGSyncService(db=session)
await sync_service.complete_sync(
sync_id=sync_log_id,
status="error",
error_message=str(exc),
)
return {"status": "error", "message": str(exc)}
@_register
async def sync_single_source_task(
ctx: dict,
source_id: str,
sync_log_id: str,
) -> dict:
"""Sync a single configured sync source.
Called by :meth:`SyncSourceService.trigger_sync`.
"""
logger.info("[WORKER] Starting source sync: source=%s, log=%s", source_id, sync_log_id)
from backend.core.database import open_session
from backend.services.sync_source import SyncSourceService
try:
async with open_session() as session:
source_service = SyncSourceService(db=session)
source = await source_service.get_source(source_id)
# For local-file connectors, sync the configured path
connector_type = source.connector_type
config = source.config if isinstance(source.config, dict) else {}
if "path" in config:
result = await sync_collection_task(
ctx,
sync_log_id=sync_log_id,
source=connector_type,
collection_name=source.collection_name,
mode=source.sync_mode,
path=config["path"],
)
await source_service.update_after_sync(
source_id=source_id,
status=result.get("status", "error"),
error=result.get("message"),
)
return result
else:
await source_service.update_after_sync(
source_id=source_id,
status="error",
error="No path configured in source config",
)
return {"status": "error", "message": "No path configured"}
except Exception as exc:
logger.exception("[WORKER] Source sync failed: source=%s", source_id)
async with open_session() as session:
source_service = SyncSourceService(db=session)
await source_service.update_after_sync(
source_id=source_id,
status="error",
error=str(exc),
)
return {"status": "error", "message": str(exc)}
+72
View File
@@ -28,6 +28,11 @@ services:
- LLM_MODEL=${LLM_MODEL:-llama}
- DATABASE_URL=${DATABASE_URL:-postgresql+asyncpg://agent_alpha:agent_alpha@postgres:5432/agent_alpha}
- VALKEY_URL=${VALKEY_URL:-redis://valkey:6379/0}
# RAG / Milvus
- MILVUS_URI=${MILVUS_URI:-http://milvus:19530}
- MILVUS_TOKEN=${MILVUS_TOKEN:-}
- MEDIA_DIR=${MEDIA_DIR:-/app/media}
- MAX_UPLOAD_SIZE_MB=${MAX_UPLOAD_SIZE_MB:-50}
restart: unless-stopped
depends_on:
@@ -35,6 +40,8 @@ services:
condition: service_healthy
valkey:
condition: service_healthy
milvus:
condition: service_healthy
# Uncomment to use a local .env file instead of env vars:
# env_file:
# - .env
@@ -89,6 +96,68 @@ services:
retries: 10
restart: unless-stopped
# ── Milvus Vector Database ──────────────────────────────────────
# Requires etcd (coordinator) and MinIO (storage).
milvus:
image: milvusdb/milvus:v2.5-latest
container_name: agent-alpha-milvus
ports:
- "19530:19530"
environment:
ETCD_ENDPOINTS: etcd:2379
MINIO_ADDRESS: minio:9000
MINIO_ACCESS_KEY: minioadmin
MINIO_SECRET_KEY: minioadmin
volumes:
- milvus_data:/var/lib/milvus
depends_on:
etcd:
condition: service_healthy
minio:
condition: service_healthy
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:9091/health"]
interval: 10s
timeout: 5s
retries: 10
restart: unless-stopped
etcd:
image: quay.io/coreos/etcd:v3.5.17
container_name: agent-alpha-etcd
environment:
ETCD_AUTO_COMPACTION_MODE: revision
ETCD_AUTO_COMPACTION_RETENTION: "1000"
ETCD_QUOTA_BACKEND_BYTES: "4294967296"
ETCD_SNAPSHOT_COUNT: "50000"
volumes:
- etcd_data:/etcd
healthcheck:
test: ["CMD", "etcdctl", "endpoint", "health"]
interval: 10s
timeout: 5s
retries: 10
restart: unless-stopped
minio:
image: minio/minio:RELEASE.2024-11-17T22-20-09Z
container_name: agent-alpha-minio
ports:
- "9000:9000"
- "9001:9001"
environment:
MINIO_ACCESS_KEY: minioadmin
MINIO_SECRET_KEY: minioadmin
volumes:
- minio_data:/data
command: server /data --console-address ":9001"
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
interval: 10s
timeout: 5s
retries: 10
restart: unless-stopped
# ── Ollama (optional — uncomment to run the LLM locally) ────────
# ollama:
# image: ollama/ollama:latest
@@ -109,4 +178,7 @@ services:
volumes:
postgres_data:
valkey_data:
milvus_data:
etcd_data:
minio_data:
# ollama_data:
+12
View File
@@ -20,4 +20,16 @@ dependencies = [
"bcrypt>=4.2.0",
"redis[hiredis]>=5.2.0",
"sqlalchemy[asyncio]>=2.0.36",
"pymupdf>=1.27.2.3",
"python-docx>=1.2.0",
"langchain-text-splitters>=1.1.2",
"sentence-transformers>=5.5.1",
# Vector Database
"pymilvus>=2.5.0",
# Async Task Queue (lightweight, redis-based)
# (using redis directly instead of arq to avoid dep conflicts)
# Embeddings
"openai>=1.0.0",
# Hybrid Search
"rank-bm25>=0.2.2",
]
Generated
+1276 -9
View File
File diff suppressed because it is too large Load Diff