diff --git a/backend/app.py b/backend/app.py index c27732e..69ddbac 100644 --- a/backend/app.py +++ b/backend/app.py @@ -24,12 +24,18 @@ logger = logging.getLogger("agent_alpha") def _configure_logging(*, debug: bool = False) -> None: """Configure standard logging to match FastAPI/uvicorn's log format.""" + level = logging.DEBUG if debug else logging.INFO handler = logging.StreamHandler() handler.setFormatter(DefaultFormatter("%(levelprefix)s %(message)s")) + # Configure the agent_alpha logger logger = logging.getLogger("agent_alpha") logger.addHandler(handler) - logger.setLevel(logging.DEBUG if debug else logging.INFO) + logger.setLevel(level) logger.propagate = False + # Ensure all backend.* loggers output to stderr at the right level + logging.getLogger("backend").addHandler(handler) + logging.getLogger("backend").setLevel(level) + logging.getLogger("backend").propagate = False _configure_logging() diff --git a/backend/repositories/rag_document_repo.py b/backend/repositories/rag_document_repo.py index cd2181e..1a52a01 100644 --- a/backend/repositories/rag_document_repo.py +++ b/backend/repositories/rag_document_repo.py @@ -5,7 +5,7 @@ from __future__ import annotations from datetime import datetime from uuid import UUID -from sqlalchemy import delete, func, select, update +from sqlalchemy import delete as sa_delete, func, select, update from sqlalchemy.ext.asyncio import AsyncSession from backend.db.models.rag_document import RAGDocument @@ -69,7 +69,7 @@ async def update_status( async def delete(session: AsyncSession, doc_id: UUID) -> None: """Delete a RAG document record.""" - stmt = delete(RAGDocument).where(RAGDocument.id == doc_id) + stmt = sa_delete(RAGDocument).where(RAGDocument.id == doc_id) await session.execute(stmt) await session.flush() @@ -77,7 +77,7 @@ async def delete(session: AsyncSession, doc_id: UUID) -> None: async def delete_by_collection(session: AsyncSession, collection_name: str) -> int: """Delete all RAG documents for a collection. Returns count.""" stmt = ( - delete(RAGDocument) + sa_delete(RAGDocument) .where(RAGDocument.collection_name == collection_name) .returning(RAGDocument.id) ) diff --git a/backend/repositories/sync_source_repo.py b/backend/repositories/sync_source_repo.py index 102de50..3eaa4a4 100644 --- a/backend/repositories/sync_source_repo.py +++ b/backend/repositories/sync_source_repo.py @@ -5,7 +5,7 @@ from __future__ import annotations from datetime import datetime from uuid import UUID -from sqlalchemy import delete, select, update +from sqlalchemy import delete as sa_delete, select, update from sqlalchemy.ext.asyncio import AsyncSession from backend.db.models.sync_source import SyncSource @@ -71,7 +71,7 @@ async def update( async def delete(session: AsyncSession, source_id: UUID) -> None: """Delete a sync source.""" - stmt = delete(SyncSource).where(SyncSource.id == source_id) + stmt = sa_delete(SyncSource).where(SyncSource.id == source_id) await session.execute(stmt) await session.flush() diff --git a/backend/routes/rag.py b/backend/routes/rag.py index 1d09c7e..c0cb28b 100644 --- a/backend/routes/rag.py +++ b/backend/routes/rag.py @@ -257,7 +257,10 @@ async def delete_document( ) -> RAGMessageResponse: """Delete a document from tracking and cascade to vector store + file storage.""" service = RAGDocumentService(db=session) - ingestion = _get_ingestion_service() + try: + ingestion = _get_ingestion_service() + except Exception: + ingestion = None await service.delete_document(doc_id, ingestion_service=ingestion) return RAGMessageResponse(message=f"Document '{doc_id}' deleted") diff --git a/backend/services/rag/documents.py b/backend/services/rag/documents.py index e9f3f61..0af7b1a 100644 --- a/backend/services/rag/documents.py +++ b/backend/services/rag/documents.py @@ -438,6 +438,12 @@ class DocumentProcessor: document = await self.pdf_parser.parse(filepath) else: raise ValueError(f"Unsupported file type: {filepath.suffix}") + logger.info( + "[DOCPROC] Parsed '%s': %d pages, %d images", + filepath.name, + len(document.pages), + sum(len(p.images) for p in document.pages), + ) # Describe images using LLM vision before chunking await self._describe_images(document) @@ -464,4 +470,12 @@ class DocumentProcessor: # Add chunked pages to original document document.chunked_pages = chunked_pages + logger.info( + "[DOCPROC] Chunked '%s': %d chunks (strategy=%s, size=%d, overlap=%d)", + filepath.name, + len(chunked_pages), + self.settings.chunking_strategy, + self.settings.chunk_size, + self.settings.chunk_overlap, + ) return document diff --git a/backend/services/rag/embeddings.py b/backend/services/rag/embeddings.py index fda41e9..ad00008 100644 --- a/backend/services/rag/embeddings.py +++ b/backend/services/rag/embeddings.py @@ -1,3 +1,5 @@ +import logging + from abc import ABC, abstractmethod from openai import OpenAI @@ -101,7 +103,19 @@ class OpenAIEmbeddingProvider(BaseEmbeddingProvider): texts = [ doc.chunk_content if doc.chunk_content else "" for doc in (document.chunked_pages or []) ] - return self.embed_queries(texts) + logging.getLogger(__name__).info( + "[EMBED] Embedding %d chunks for doc '%s' (model=%s)", + len(texts), + document.metadata.filename, + self.model, + ) + vectors = self.embed_queries(texts) + logging.getLogger(__name__).info( + "[EMBED] Got %d vectors, first vector len=%d", + len(vectors), + len(vectors[0]) if vectors else 0, + ) + return vectors def warmup(self) -> None: """Warmup method for OpenAI client. @@ -160,6 +174,13 @@ class EmbeddingService: Returns: List of embedding vectors for each chunk. """ + chunk_count = len(document.chunked_pages or []) + logging.getLogger(__name__).info( + "[EMBEDSVC] Embedding doc '%s': %d chunks, expected_dim=%d", + document.metadata.filename, + chunk_count, + self.expected_dim, + ) results = self.provider.embed_document(document) if results and len(results[0]) != self.expected_dim: raise ValueError( diff --git a/backend/services/rag/ingestion.py b/backend/services/rag/ingestion.py index a7fd136..2d28354 100644 --- a/backend/services/rag/ingestion.py +++ b/backend/services/rag/ingestion.py @@ -100,6 +100,13 @@ class IngestionService: source_path: Override source path (e.g., gdrive://id, s3://bucket/key). """ try: + logger.info( + "[INGEST] Starting ingestion: file='%s', collection='%s', replace=%s, source='%s'", + filepath.name, + collection_name, + replace, + source_path or filepath.name, + ) # Processing (Parsing + Chunking) document: Document = await self.processor.process_file(filepath) @@ -127,6 +134,12 @@ class IngestionService: await self.store.delete_document(collection_name, existing_id) logger.info(f"Replaced existing document {existing_id} for '{filepath.name}'") + logger.info( + "[INGEST] Inserting into vector store: collection='%s', doc_id=%s, chunks=%d", + collection_name, + document.id, + len(document.chunked_pages or []), + ) # Storage (Embedding + Insertion) await self.store.insert_document( collection_name=collection_name, @@ -147,6 +160,12 @@ class IngestionService: }, ) + logger.info( + "[INGEST] Success: action=%s, doc_id=%s, chunks=%d", + action, + document.id, + len(document.chunked_pages or []), + ) return IngestionResult( status=IngestionStatus.DONE, document_id=document.id, @@ -154,7 +173,7 @@ class IngestionService: ) except Exception as e: - logger.error(f"Ingestion error for {filepath.name}: {e!s}") + logger.error(f"[INGEST] Error for {filepath.name}: {e!s}") return IngestionResult( status=IngestionStatus.ERROR, error_message=str(e), diff --git a/backend/services/rag/vectorstore.py b/backend/services/rag/vectorstore.py index f7faf83..e426471 100644 --- a/backend/services/rag/vectorstore.py +++ b/backend/services/rag/vectorstore.py @@ -180,6 +180,13 @@ class MilvusVectorStore(BaseVectorStore): await self._ensure_collection(collection_name) if not document.chunked_pages: raise ValueError("Document has no chunked pages.") + logger.info( + "[MILVUS] Inserting doc '%s' into collection '%s': %d chunks, dim=%d", + document.metadata.filename, + collection_name, + len(document.chunked_pages), + self.settings.embeddings_config.dim, + ) vectors = self.embedder.embed_document(document) data = [ { @@ -191,7 +198,11 @@ class MilvusVectorStore(BaseVectorStore): } for i, chunk in enumerate(document.chunked_pages) ] - await self.client.insert(collection_name, data=data) + insert_result = await self.client.insert(collection_name, data=data) + logger.info( + "[MILVUS] Insert result: %s", + insert_result, + ) async def search( self, collection_name: str, query: str, limit: int = 4, filter: str = "" diff --git a/backend/services/rag_document.py b/backend/services/rag_document.py index 7631342..eb351f2 100644 --- a/backend/services/rag_document.py +++ b/backend/services/rag_document.py @@ -113,6 +113,13 @@ class RAGDocumentService: allowed = get_supported_formats(getattr(settings, "pdf_parser", "pymupdf")) max_size = settings.max_upload_size_mb * 1024 * 1024 + logger.info( + "[RAGDOC] dispatch_upload: filename='%s', size=%d, collection='%s', replace=%s", + filename, + len(file_data), + collection_name, + replace, + ) ext = Path(filename).suffix.lower() if ext not in allowed: raise BadRequestError( @@ -135,9 +142,11 @@ class RAGDocumentService: storage_path=storage_path, ) doc_id = rag_doc.id + logger.info("[RAGDOC] Saved file to '%s', doc_id=%s", storage_path, doc_id) # Ensure the target collection exists before the worker starts await vector_store.create_collection(collection_name) + logger.info("[RAGDOC] Ensured collection '%s' exists", collection_name) # Stage the upload in the volume shared with the worker container tmp_dir = os.path.join(str(settings.media_dir), "_rag_tmp")