mirror of
https://github.com/furyhawk/agent_alpha.git
synced 2026-07-21 02:25:34 +00:00
feat: Enhance logging across services for better traceability and debugging
This commit is contained in:
+7
-1
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
)
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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")
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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 = ""
|
||||
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user