diff --git a/trustgraph-flow/trustgraph/chunking/recursive/chunker.py b/trustgraph-flow/trustgraph/chunking/recursive/chunker.py index aa48cc57..fe182b14 100755 --- a/trustgraph-flow/trustgraph/chunking/recursive/chunker.py +++ b/trustgraph-flow/trustgraph/chunking/recursive/chunker.py @@ -4,12 +4,16 @@ Simple decoder, accepts text documents on input, outputs chunks from the as text as separate output objects. """ +import logging from langchain_text_splitters import RecursiveCharacterTextSplitter from prometheus_client import Histogram from ... schema import TextDocument, Chunk from ... base import FlowProcessor, ConsumerSpec, ProducerSpec +# Module logger +logger = logging.getLogger(__name__) + default_ident = "chunker" class Processor(FlowProcessor): @@ -54,12 +58,12 @@ class Processor(FlowProcessor): ) ) - print("Chunker initialised", flush=True) + logger.info("Recursive chunker initialized") async def on_message(self, msg, consumer, flow): v = msg.value() - print(f"Chunking {v.metadata.id}...", flush=True) + logger.info(f"Chunking document {v.metadata.id}...") texts = self.text_splitter.create_documents( [v.text.decode("utf-8")] @@ -67,7 +71,7 @@ class Processor(FlowProcessor): for ix, chunk in enumerate(texts): - print("Chunk", len(chunk.page_content), flush=True) + logger.debug(f"Created chunk of size {len(chunk.page_content)}") r = Chunk( metadata=v.metadata, @@ -80,7 +84,7 @@ class Processor(FlowProcessor): await flow("output").send(r) - print("Done.", flush=True) + logger.debug("Document chunking complete") @staticmethod def add_args(parser): diff --git a/trustgraph-flow/trustgraph/chunking/token/chunker.py b/trustgraph-flow/trustgraph/chunking/token/chunker.py index ff217350..028f62fa 100755 --- a/trustgraph-flow/trustgraph/chunking/token/chunker.py +++ b/trustgraph-flow/trustgraph/chunking/token/chunker.py @@ -4,12 +4,16 @@ Simple decoder, accepts text documents on input, outputs chunks from the as text as separate output objects. """ +import logging from langchain_text_splitters import TokenTextSplitter from prometheus_client import Histogram from ... schema import TextDocument, Chunk from ... base import FlowProcessor +# Module logger +logger = logging.getLogger(__name__) + default_ident = "chunker" class Processor(FlowProcessor): @@ -53,12 +57,12 @@ class Processor(FlowProcessor): ) ) - print("Chunker initialised", flush=True) + logger.info("Token chunker initialized") async def on_message(self, msg, consumer, flow): v = msg.value() - print(f"Chunking {v.metadata.id}...", flush=True) + logger.info(f"Chunking document {v.metadata.id}...") texts = self.text_splitter.create_documents( [v.text.decode("utf-8")] @@ -66,7 +70,7 @@ class Processor(FlowProcessor): for ix, chunk in enumerate(texts): - print("Chunk", len(chunk.page_content), flush=True) + logger.debug(f"Created chunk of size {len(chunk.page_content)}") r = Chunk( metadata=v.metadata, @@ -79,7 +83,7 @@ class Processor(FlowProcessor): await flow("output").send(r) - print("Done.", flush=True) + logger.debug("Document chunking complete") @staticmethod def add_args(parser): diff --git a/trustgraph-flow/trustgraph/config/service/config.py b/trustgraph-flow/trustgraph/config/service/config.py index de684ec2..c9d315b0 100644 --- a/trustgraph-flow/trustgraph/config/service/config.py +++ b/trustgraph-flow/trustgraph/config/service/config.py @@ -1,9 +1,14 @@ +import logging + from trustgraph.schema import ConfigResponse from trustgraph.schema import ConfigValue, Error from ... tables.config import ConfigTableStore +# Module logger +logger = logging.getLogger(__name__) + class ConfigurationClass: async def keys(self): @@ -228,7 +233,7 @@ class Configuration: async def handle(self, msg): - print("Handle message ", msg.operation) + logger.debug(f"Handling config message: {msg.operation}") if msg.operation == "get": diff --git a/trustgraph-flow/trustgraph/config/service/flow.py b/trustgraph-flow/trustgraph/config/service/flow.py index 83e6835e..3e83f8fa 100644 --- a/trustgraph-flow/trustgraph/config/service/flow.py +++ b/trustgraph-flow/trustgraph/config/service/flow.py @@ -1,6 +1,10 @@ from trustgraph.schema import FlowResponse, Error import json +import logging + +# Module logger +logger = logging.getLogger(__name__) class FlowConfig: def __init__(self, config): @@ -41,7 +45,7 @@ class FlowConfig: async def handle_delete_class(self, msg): - print(msg) + logger.debug(f"Flow config message: {msg}") await self.config.get("flow-classes").delete(msg.class_name) @@ -218,7 +222,7 @@ class FlowConfig: async def handle(self, msg): - print("Handle message ", msg.operation) + logger.debug(f"Handling flow message: {msg.operation}") if msg.operation == "list-classes": resp = await self.handle_list_classes(msg) diff --git a/trustgraph-flow/trustgraph/config/service/service.py b/trustgraph-flow/trustgraph/config/service/service.py index 1ef81341..8c20e268 100644 --- a/trustgraph-flow/trustgraph/config/service/service.py +++ b/trustgraph-flow/trustgraph/config/service/service.py @@ -3,6 +3,8 @@ Config service. Manages system global configuration state """ +import logging + from trustgraph.schema import Error from trustgraph.schema import ConfigRequest, ConfigResponse, ConfigPush @@ -20,6 +22,9 @@ from . flow import FlowConfig from ... base import ProcessorMetrics, ConsumerMetrics, ProducerMetrics from ... base import Consumer, Producer +# Module logger +logger = logging.getLogger(__name__) + # FIXME: How to ensure this doesn't conflict with other usage? keyspace = "config" @@ -146,7 +151,7 @@ class Processor(AsyncProcessor): self.flow = FlowConfig(self.config) - print("Service initialised.") + logger.info("Config service initialized") async def start(self): @@ -172,7 +177,7 @@ class Processor(AsyncProcessor): # Race condition, should make sure version & config sync - print("Pushed version ", await self.config.get_version()) + logger.info(f"Pushed configuration version {await self.config.get_version()}") async def on_config_request(self, msg, consumer, flow): @@ -183,7 +188,7 @@ class Processor(AsyncProcessor): # Sender-produced ID id = msg.properties()["id"] - print(f"Handling {id}...", flush=True) + logger.info(f"Handling config request {id}...") resp = await self.config.handle(v) @@ -214,7 +219,7 @@ class Processor(AsyncProcessor): # Sender-produced ID id = msg.properties()["id"] - print(f"Handling {id}...", flush=True) + logger.info(f"Handling flow request {id}...") resp = await self.flow.handle(v) diff --git a/trustgraph-flow/trustgraph/cores/knowledge.py b/trustgraph-flow/trustgraph/cores/knowledge.py index 8c082601..898e8e15 100644 --- a/trustgraph-flow/trustgraph/cores/knowledge.py +++ b/trustgraph-flow/trustgraph/cores/knowledge.py @@ -8,6 +8,10 @@ from .. base import Publisher import base64 import asyncio import uuid +import logging + +# Module logger +logger = logging.getLogger(__name__) class KnowledgeManager: @@ -26,7 +30,7 @@ class KnowledgeManager: async def delete_kg_core(self, request, respond): - print("Deleting core...", flush=True) + logger.info("Deleting knowledge core...") await self.table_store.delete_kg_core( request.user, request.id @@ -44,7 +48,7 @@ class KnowledgeManager: async def get_kg_core(self, request, respond): - print("Get core...", flush=True) + logger.info("Getting knowledge core...") async def publish_triples(t): await respond( @@ -82,7 +86,7 @@ class KnowledgeManager: publish_ge, ) - print("Get complete", flush=True) + logger.debug("Knowledge core retrieval complete") await respond( KnowledgeResponse( @@ -158,13 +162,13 @@ class KnowledgeManager: async def core_loader(self): - print("Running...", flush=True) + logger.info("Knowledge background processor running...") while True: - print("Wait for next load...", flush=True) + logger.debug("Waiting for next load...") request, respond = await self.loader_queue.get() - print("Loading...", request.id, flush=True) + logger.info(f"Loading knowledge: {request.id}") try: @@ -204,7 +208,7 @@ class KnowledgeManager: except Exception as e: - print("Exception:", e, flush=True) + logger.error(f"Knowledge exception: {e}", exc_info=True) await respond( KnowledgeResponse( error = Error( @@ -219,15 +223,15 @@ class KnowledgeManager: ) - print("Going to start loading...", flush=True) + logger.debug("Starting knowledge loading process...") try: t_pub = None ge_pub = None - print(t_q, flush=True) - print(ge_q, flush=True) + logger.debug(f"Triples queue: {t_q}") + logger.debug(f"Graph embeddings queue: {ge_q}") t_pub = Publisher( self.flow_config.pulsar_client, t_q, @@ -238,7 +242,7 @@ class KnowledgeManager: schema=GraphEmbeddings ) - print("Start publishers...", flush=True) + logger.debug("Starting publishers...") await t_pub.start() await ge_pub.start() @@ -246,7 +250,7 @@ class KnowledgeManager: async def publish_triples(t): await t_pub.send(None, t) - print("Publish triples...", flush=True) + logger.debug("Publishing triples...") # Remove doc table row await self.table_store.get_triples( @@ -258,7 +262,7 @@ class KnowledgeManager: async def publish_ge(g): await ge_pub.send(None, g) - print("Publish GEs...", flush=True) + logger.debug("Publishing graph embeddings...") # Remove doc table row await self.table_store.get_graph_embeddings( @@ -267,19 +271,19 @@ class KnowledgeManager: publish_ge, ) - print("Completed that.", flush=True) + logger.debug("Knowledge loading completed") except Exception as e: - print("Exception:", e, flush=True) + logger.error(f"Knowledge exception: {e}", exc_info=True) finally: - print("Stopping publishers...", flush=True) + logger.debug("Stopping publishers...") if t_pub: await t_pub.stop() if ge_pub: await ge_pub.stop() - print("Done", flush=True) + logger.debug("Knowledge processing done") continue diff --git a/trustgraph-flow/trustgraph/cores/service.py b/trustgraph-flow/trustgraph/cores/service.py index 810d159d..ade3d12c 100755 --- a/trustgraph-flow/trustgraph/cores/service.py +++ b/trustgraph-flow/trustgraph/cores/service.py @@ -7,6 +7,7 @@ from functools import partial import asyncio import base64 import json +import logging from .. base import AsyncProcessor, Consumer, Producer, Publisher, Subscriber from .. base import ConsumerMetrics, ProducerMetrics @@ -21,6 +22,9 @@ from .. exceptions import RequestError from . knowledge import KnowledgeManager +# Module logger +logger = logging.getLogger(__name__) + default_ident = "knowledge" default_knowledge_request_queue = knowledge_request_queue @@ -96,7 +100,7 @@ class Processor(AsyncProcessor): self.flows = {} - print("Initialised.", flush=True) + logger.info("Knowledge service initialized") async def start(self): @@ -106,7 +110,7 @@ class Processor(AsyncProcessor): async def on_knowledge_config(self, config, version): - print("config version", version) + logger.info(f"Configuration version: {version}") if "flows" in config: @@ -115,14 +119,14 @@ class Processor(AsyncProcessor): for k, v in config["flows"].items() } - print(self.flows) + logger.debug(f"Flows: {self.flows}") async def process_request(self, v, id): if v.operation is None: raise RequestError("Null operation") - print("request", v.operation) + logger.debug(f"Knowledge request: {v.operation}") impls = { "list-kg-cores": self.knowledge.list_kg_cores, @@ -150,7 +154,7 @@ class Processor(AsyncProcessor): id = msg.properties()["id"] - print(f"Handling input {id}...", flush=True) + logger.info(f"Handling knowledge input {id}...") try: @@ -187,7 +191,7 @@ class Processor(AsyncProcessor): return - print("Done.", flush=True) + logger.debug("Knowledge input processing complete") @staticmethod def add_args(parser): diff --git a/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py b/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py index 3f836832..bb641a26 100755 --- a/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py +++ b/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py @@ -6,11 +6,15 @@ PDF document as text as separate output objects. import tempfile import base64 +import logging from langchain_community.document_loaders import PyPDFLoader from ... schema import Document, TextDocument, Metadata from ... base import FlowProcessor, ConsumerSpec, ProducerSpec +# Module logger +logger = logging.getLogger(__name__) + default_ident = "pdf-decoder" class Processor(FlowProcessor): @@ -40,15 +44,15 @@ class Processor(FlowProcessor): ) ) - print("PDF inited", flush=True) + logger.info("PDF decoder initialized") async def on_message(self, msg, consumer, flow): - print("PDF message received", flush=True) + logger.debug("PDF message received") v = msg.value() - print(f"Decoding {v.metadata.id}...", flush=True) + logger.info(f"Decoding PDF {v.metadata.id}...") with tempfile.NamedTemporaryFile(delete_on_close=False) as fp: @@ -62,7 +66,7 @@ class Processor(FlowProcessor): for ix, page in enumerate(pages): - print("page", ix, flush=True) + logger.debug(f"Processing page {ix}") r = TextDocument( metadata=v.metadata, @@ -71,7 +75,7 @@ class Processor(FlowProcessor): await flow("output").send(r) - print("Done.", flush=True) + logger.debug("PDF decoding complete") @staticmethod def add_args(parser): diff --git a/trustgraph-flow/trustgraph/embeddings/fastembed/processor.py b/trustgraph-flow/trustgraph/embeddings/fastembed/processor.py index a4ae35dc..0357e4a3 100755 --- a/trustgraph-flow/trustgraph/embeddings/fastembed/processor.py +++ b/trustgraph-flow/trustgraph/embeddings/fastembed/processor.py @@ -4,10 +4,15 @@ Embeddings service, applies an embeddings model using fastembed Input is text, output is embeddings vector. """ +import logging + from ... base import EmbeddingsService from fastembed import TextEmbedding +# Module logger +logger = logging.getLogger(__name__) + default_ident = "embeddings" default_model="sentence-transformers/all-MiniLM-L6-v2" @@ -22,7 +27,7 @@ class Processor(EmbeddingsService): **params | { "model": model } ) - print("Get model...", flush=True) + logger.info("Loading FastEmbed model...") self.embeddings = TextEmbedding(model_name = model) async def on_embeddings(self, text): diff --git a/trustgraph-vertexai/trustgraph/model/text_completion/vertexai/llm.py b/trustgraph-vertexai/trustgraph/model/text_completion/vertexai/llm.py index c6d869e6..24cc576c 100755 --- a/trustgraph-vertexai/trustgraph/model/text_completion/vertexai/llm.py +++ b/trustgraph-vertexai/trustgraph/model/text_completion/vertexai/llm.py @@ -19,6 +19,7 @@ Google Cloud. Input is prompt, output is response. from google.oauth2 import service_account import google import vertexai +import logging # Why is preview here? from vertexai.generative_models import ( @@ -29,6 +30,9 @@ from vertexai.generative_models import ( from .... exceptions import TooManyRequests from .... base import LlmService, LlmResult +# Module logger +logger = logging.getLogger(__name__) + default_ident = "text-completion" default_model = 'gemini-2.0-flash-001' @@ -91,7 +95,7 @@ class Processor(LlmService): ), ] - print("Initialise VertexAI...", flush=True) + logger.info("Initializing VertexAI...") if private_key: credentials = ( @@ -113,11 +117,11 @@ class Processor(LlmService): location=region ) - print(f"Initialise model {model}", flush=True) + logger.info(f"Initializing model {model}") self.llm = GenerativeModel(model) self.model = model - print("Initialisation complete", flush=True) + logger.info("VertexAI initialization complete") async def generate_content(self, system, prompt): @@ -137,16 +141,16 @@ class Processor(LlmService): model = self.model ) - print(f"Input Tokens: {resp.in_token}", flush=True) - print(f"Output Tokens: {resp.out_token}", flush=True) + logger.info(f"Input Tokens: {resp.in_token}") + logger.info(f"Output Tokens: {resp.out_token}") - print("Send response...", flush=True) + logger.debug("Send response...") return resp except google.api_core.exceptions.ResourceExhausted as e: - print("Hit rate limit:", e, flush=True) + logger.warning(f"Hit rate limit: {e}") # Leave rate limit retries to the base handler raise TooManyRequests() @@ -154,7 +158,7 @@ class Processor(LlmService): except Exception as e: # Apart from rate limits, treat all exceptions as unrecoverable - print(f"Exception: {e}") + logger.error(f"VertexAI LLM exception: {e}", exc_info=True) raise e @staticmethod