From 4b50abbf5a06688c7ac1884b43c20070a493f819 Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Wed, 30 Jul 2025 22:21:03 +0100 Subject: [PATCH] Logging strategy updates --- .../trustgraph/librarian/blob_store.py | 14 +++++--- .../trustgraph/librarian/librarian.py | 32 +++++++++-------- .../retrieval/document_rag/document_rag.py | 26 ++++++++------ .../trustgraph/retrieval/document_rag/rag.py | 12 ++++--- .../retrieval/graph_rag/graph_rag.py | 34 +++++++++++-------- .../trustgraph/retrieval/graph_rag/rag.py | 12 ++++--- 6 files changed, 77 insertions(+), 53 deletions(-) diff --git a/trustgraph-flow/trustgraph/librarian/blob_store.py b/trustgraph-flow/trustgraph/librarian/blob_store.py index 3368f57e..2a71f5a8 100644 --- a/trustgraph-flow/trustgraph/librarian/blob_store.py +++ b/trustgraph-flow/trustgraph/librarian/blob_store.py @@ -5,6 +5,10 @@ from .. exceptions import RequestError from minio import Minio import time import io +import logging + +# Module logger +logger = logging.getLogger(__name__) class BlobStore: @@ -23,7 +27,7 @@ class BlobStore: self.bucket_name = bucket_name - print("Connected to minio", flush=True) + logger.info("Connected to MinIO") self.ensure_bucket() @@ -33,9 +37,9 @@ class BlobStore: found = self.minio.bucket_exists(self.bucket_name) if not found: self.minio.make_bucket(self.bucket_name) - print("Created bucket", self.bucket_name, flush=True) + logger.info(f"Created bucket {self.bucket_name}") else: - print("Bucket", self.bucket_name, "already exists", flush=True) + logger.debug(f"Bucket {self.bucket_name} already exists") async def add(self, object_id, blob, kind): @@ -48,7 +52,7 @@ class BlobStore: content_type = kind, ) - print("Add blob complete", flush=True) + logger.debug("Add blob complete") async def remove(self, object_id): @@ -58,7 +62,7 @@ class BlobStore: object_name = "doc/" + str(object_id), ) - print("Remove blob complete", flush=True) + logger.debug("Remove blob complete") async def get(self, object_id): diff --git a/trustgraph-flow/trustgraph/librarian/librarian.py b/trustgraph-flow/trustgraph/librarian/librarian.py index 87f1d27f..59a71f48 100644 --- a/trustgraph-flow/trustgraph/librarian/librarian.py +++ b/trustgraph-flow/trustgraph/librarian/librarian.py @@ -5,9 +5,13 @@ from .. exceptions import RequestError from .. tables.library import LibraryTableStore from . blob_store import BlobStore import base64 +import logging import uuid +# Module logger +logger = logging.getLogger(__name__) + class Librarian: def __init__( @@ -70,7 +74,7 @@ class Librarian: async def remove_document(self, request): - print("Removing doc...") + logger.debug("Removing document...") if not await self.table_store.document_exists( request.user, @@ -92,7 +96,7 @@ class Librarian: request.document_id ) - print("Remove complete", flush=True) + logger.debug("Remove complete") return LibrarianResponse( error = None, @@ -104,7 +108,7 @@ class Librarian: async def update_document(self, request): - print("Updating doc...") + logger.debug("Updating document...") # You can't update the document ID, user or kind. @@ -116,7 +120,7 @@ class Librarian: await self.table_store.update_document(request.document_metadata) - print("Update complete", flush=True) + logger.debug("Update complete") return LibrarianResponse( error = None, @@ -128,14 +132,14 @@ class Librarian: async def get_document_metadata(self, request): - print("Get doc...") + logger.debug("Getting document metadata...") doc = await self.table_store.get_document( request.user, request.document_id ) - print("Get complete", flush=True) + logger.debug("Get complete") return LibrarianResponse( error = None, @@ -147,7 +151,7 @@ class Librarian: async def get_document_content(self, request): - print("Get doc content...") + logger.debug("Getting document content...") object_id = await self.table_store.get_document_object_id( request.user, @@ -158,7 +162,7 @@ class Librarian: object_id ) - print("Get complete", flush=True) + logger.debug("Get complete") return LibrarianResponse( error = None, @@ -170,7 +174,7 @@ class Librarian: async def add_processing(self, request): - print("Add processing") + logger.debug("Adding processing metadata...") if await self.table_store.processing_exists( request.processing_metadata.user, @@ -192,13 +196,13 @@ class Librarian: object_id ) - print("Got content") + logger.debug("Retrieved content") - print("Add processing...") + logger.debug("Adding processing to table...") await self.table_store.add_processing(request.processing_metadata) - print("Invoke document processing...") + logger.debug("Invoking document processing...") await self.load_document( document = doc, @@ -218,7 +222,7 @@ class Librarian: async def remove_processing(self, request): - print("Removing processing...") + logger.debug("Removing processing metadata...") if not await self.table_store.processing_exists( request.user, @@ -232,7 +236,7 @@ class Librarian: request.processing_id ) - print("Remove complete", flush=True) + logger.debug("Remove complete") return LibrarianResponse( error = None, diff --git a/trustgraph-flow/trustgraph/retrieval/document_rag/document_rag.py b/trustgraph-flow/trustgraph/retrieval/document_rag/document_rag.py index 5e3c9b41..d885757e 100644 --- a/trustgraph-flow/trustgraph/retrieval/document_rag/document_rag.py +++ b/trustgraph-flow/trustgraph/retrieval/document_rag/document_rag.py @@ -1,5 +1,9 @@ import asyncio +import logging + +# Module logger +logger = logging.getLogger(__name__) LABEL="http://www.w3.org/2000/01/rdf-schema#label" @@ -18,12 +22,12 @@ class Query: async def get_vector(self, query): if self.verbose: - print("Compute embeddings...", flush=True) + logger.debug("Computing embeddings...") qembeds = await self.rag.embeddings_client.embed(query) if self.verbose: - print("Done.", flush=True) + logger.debug("Embeddings computed") return qembeds @@ -32,7 +36,7 @@ class Query: vectors = await self.get_vector(query) if self.verbose: - print("Get docs...", flush=True) + logger.debug("Getting documents...") docs = await self.rag.doc_embeddings_client.query( vectors, limit=self.doc_limit, @@ -40,9 +44,9 @@ class Query: ) if self.verbose: - print("Docs:", flush=True) + logger.debug("Documents:") for doc in docs: - print(doc, flush=True) + logger.debug(f" {doc}") return docs @@ -60,7 +64,7 @@ class DocumentRag: self.doc_embeddings_client = doc_embeddings_client if self.verbose: - print("Initialised", flush=True) + logger.debug("DocumentRag initialized") async def query( self, query, user="trustgraph", collection="default", @@ -68,7 +72,7 @@ class DocumentRag: ): if self.verbose: - print("Construct prompt...", flush=True) + logger.debug("Constructing prompt...") q = Query( rag=self, user=user, collection=collection, verbose=self.verbose, @@ -78,9 +82,9 @@ class DocumentRag: docs = await q.get_docs(query) if self.verbose: - print("Invoke LLM...", flush=True) - print(docs) - print(query) + logger.debug("Invoking LLM...") + logger.debug(f"Documents: {docs}") + logger.debug(f"Query: {query}") resp = await self.prompt_client.document_prompt( query = query, @@ -88,7 +92,7 @@ class DocumentRag: ) if self.verbose: - print("Done", flush=True) + logger.debug("Query processing complete") return resp diff --git a/trustgraph-flow/trustgraph/retrieval/document_rag/rag.py b/trustgraph-flow/trustgraph/retrieval/document_rag/rag.py index 8c478874..0cca2cff 100755 --- a/trustgraph-flow/trustgraph/retrieval/document_rag/rag.py +++ b/trustgraph-flow/trustgraph/retrieval/document_rag/rag.py @@ -4,12 +4,16 @@ Simple RAG service, performs query using document RAG an LLM. Input is query, output is response. """ +import logging from ... schema import DocumentRagQuery, DocumentRagResponse, Error from . document_rag import DocumentRag from ... base import FlowProcessor, ConsumerSpec, ProducerSpec from ... base import PromptClientSpec, EmbeddingsClientSpec from ... base import DocumentEmbeddingsClientSpec +# Module logger +logger = logging.getLogger(__name__) + default_ident = "document-rag" class Processor(FlowProcessor): @@ -81,7 +85,7 @@ class Processor(FlowProcessor): # Sender-produced ID id = msg.properties()["id"] - print(f"Handling input {id}...", flush=True) + logger.info(f"Handling input {id}...") if v.doc_limit: doc_limit = v.doc_limit @@ -98,13 +102,13 @@ class Processor(FlowProcessor): properties = {"id": id} ) - print("Done.", flush=True) + logger.info("Request processing complete") except Exception as e: - print(f"Exception: {e}") + logger.error(f"Document RAG service exception: {e}", exc_info=True) - print("Send error response...", flush=True) + logger.debug("Sending error response...") await flow("response").send( DocumentRagResponse( diff --git a/trustgraph-flow/trustgraph/retrieval/graph_rag/graph_rag.py b/trustgraph-flow/trustgraph/retrieval/graph_rag/graph_rag.py index 6879023a..a8b6b244 100644 --- a/trustgraph-flow/trustgraph/retrieval/graph_rag/graph_rag.py +++ b/trustgraph-flow/trustgraph/retrieval/graph_rag/graph_rag.py @@ -1,5 +1,9 @@ import asyncio +import logging + +# Module logger +logger = logging.getLogger(__name__) LABEL="http://www.w3.org/2000/01/rdf-schema#label" @@ -22,12 +26,12 @@ class Query: async def get_vector(self, query): if self.verbose: - print("Compute embeddings...", flush=True) + logger.debug("Computing embeddings...") qembeds = await self.rag.embeddings_client.embed(query) if self.verbose: - print("Done.", flush=True) + logger.debug("Done.") return qembeds @@ -36,7 +40,7 @@ class Query: vectors = await self.get_vector(query) if self.verbose: - print("Get entities...", flush=True) + logger.debug("Getting entities...") entities = await self.rag.graph_embeddings_client.query( vectors=vectors, limit=self.entity_limit, @@ -49,9 +53,9 @@ class Query: ] if self.verbose: - print("Entities:", flush=True) + logger.debug("Entities:") for ent in entities: - print(" ", ent, flush=True) + logger.debug(f" {ent}") return entities @@ -126,7 +130,7 @@ class Query: entities = await self.get_entities(query) if self.verbose: - print("Get subgraph...", flush=True) + logger.debug("Getting subgraph...") subgraph = set() @@ -157,12 +161,12 @@ class Query: sg2 = sg2[0:self.max_subgraph_size] if self.verbose: - print("Subgraph:", flush=True) + logger.debug("Subgraph:") for edge in sg2: - print(" ", str(edge), flush=True) + logger.debug(f" {str(edge)}") if self.verbose: - print("Done.", flush=True) + logger.debug("Done.") return sg2 @@ -183,7 +187,7 @@ class GraphRag: self.label_cache = {} if self.verbose: - print("Initialised", flush=True) + logger.debug("GraphRag initialized") async def query( self, query, user = "trustgraph", collection = "default", @@ -192,7 +196,7 @@ class GraphRag: ): if self.verbose: - print("Construct prompt...", flush=True) + logger.debug("Constructing prompt...") q = Query( rag = self, user = user, collection = collection, @@ -205,14 +209,14 @@ class GraphRag: kg = await q.get_labelgraph(query) if self.verbose: - print("Invoke LLM...", flush=True) - print(kg) - print(query) + logger.debug("Invoking LLM...") + logger.debug(f"Knowledge graph: {kg}") + logger.debug(f"Query: {query}") resp = await self.prompt_client.kg_prompt(query, kg) if self.verbose: - print("Done", flush=True) + logger.debug("Query processing complete") return resp diff --git a/trustgraph-flow/trustgraph/retrieval/graph_rag/rag.py b/trustgraph-flow/trustgraph/retrieval/graph_rag/rag.py index 328ae3f9..4d7b1821 100755 --- a/trustgraph-flow/trustgraph/retrieval/graph_rag/rag.py +++ b/trustgraph-flow/trustgraph/retrieval/graph_rag/rag.py @@ -4,12 +4,16 @@ Simple RAG service, performs query using graph RAG an LLM. Input is query, output is response. """ +import logging from ... schema import GraphRagQuery, GraphRagResponse, Error from . graph_rag import GraphRag from ... base import FlowProcessor, ConsumerSpec, ProducerSpec from ... base import PromptClientSpec, EmbeddingsClientSpec from ... base import GraphEmbeddingsClientSpec, TriplesClientSpec +# Module logger +logger = logging.getLogger(__name__) + default_ident = "graph-rag" default_concurrency = 1 @@ -102,7 +106,7 @@ class Processor(FlowProcessor): # Sender-produced ID id = msg.properties()["id"] - print(f"Handling input {id}...", flush=True) + logger.info(f"Handling input {id}...") if v.entity_limit: entity_limit = v.entity_limit @@ -139,13 +143,13 @@ class Processor(FlowProcessor): properties = {"id": id} ) - print("Done.", flush=True) + logger.info("Request processing complete") except Exception as e: - print(f"Exception: {e}") + logger.error(f"Graph RAG service exception: {e}", exc_info=True) - print("Send error response...", flush=True) + logger.debug("Sending error response...") await flow("response").send( GraphRagResponse(