Logging strategy updates

trustgraph-flow/trustgraph/extract/kg/agent/extract.py

@@ -1,6 +1,7 @@
 import re
 import json
 import urllib.parse
+import logging
 
 from ....schema import Chunk, Triple, Triples, Metadata, Value
 from ....schema import EntityContext, EntityContexts
@@ -12,6 +13,9 @@ from ....base import AgentClientSpec
 
 from ....template import PromptManager
 
+# Module logger
+logger = logging.getLogger(__name__)
+
 default_ident = "kg-extract-agent"
 default_concurrency = 1
 default_template_id = "agent-kg-extract"
@@ -74,10 +78,10 @@ class Processor(FlowProcessor):
 
     async def on_prompt_config(self, config, version):
 
-        print("Loading configuration version", version, flush=True)
+        logger.info(f"Loading configuration version {version}")
 
         if self.config_key not in config:
-            print(f"No key {self.config_key} in config", flush=True)
+            logger.warning(f"No key {self.config_key} in config")
             return
 
         config = config[self.config_key]
@@ -86,12 +90,12 @@ class Processor(FlowProcessor):
 
             self.manager.load_config(config)
 
-            print("Prompt configuration reloaded.", flush=True)
+            logger.info("Prompt configuration reloaded")
 
         except Exception as e:
 
-            print("Exception:", e, flush=True)
+            logger.error(f"Configuration reload exception: {e}", exc_info=True)
-            print("Configuration reload failed", flush=True)
+            logger.error("Configuration reload failed")
 
     def to_uri(self, text):
         return TRUSTGRAPH_ENTITIES + urllib.parse.quote(text)
@@ -142,7 +146,7 @@ class Processor(FlowProcessor):
             # Extract chunk text
             chunk_text = v.chunk.decode('utf-8')
 
-            print("Got chunk", flush=True)
+            logger.debug("Processing chunk for agent extraction")
 
             prompt = self.manager.render(
                 self.template_id,
@@ -151,11 +155,11 @@ class Processor(FlowProcessor):
                 }
             )
 
-            print("Prompt:", prompt, flush=True)
+            logger.debug(f"Agent prompt: {prompt}")
 
             async def handle(response):
 
-                print("Response:", response, flush=True)
+                logger.debug(f"Agent response: {response}")
 
                 if response.error is not None:
                     if response.error.message:
@@ -201,7 +205,7 @@ class Processor(FlowProcessor):
                 )
 
         except Exception as e:
-            print(f"Error processing chunk: {e}", flush=True)
+            logger.error(f"Error processing chunk: {e}", exc_info=True)
             raise
 
     def process_extraction_data(self, data, metadata):

trustgraph-flow/trustgraph/extract/kg/definitions/extract.py

@@ -7,8 +7,12 @@ entity/context definitions for embedding.
 
 import json
 import urllib.parse
+import logging
 
 from .... schema import Chunk, Triple, Triples, Metadata, Value
+
+# Module logger
+logger = logging.getLogger(__name__)
 from .... schema import EntityContext, EntityContexts
 from .... schema import PromptRequest, PromptResponse
 from .... rdf import TRUSTGRAPH_ENTITIES, DEFINITION, RDF_LABEL, SUBJECT_OF
@@ -94,11 +98,11 @@ class Processor(FlowProcessor):
     async def on_message(self, msg, consumer, flow):
 
         v = msg.value()
-        print(f"Indexing {v.metadata.id}...", flush=True)
+        logger.info(f"Extracting definitions from {v.metadata.id}...")
 
         chunk = v.chunk.decode("utf-8")
 
-        print(chunk, flush=True)
+        logger.debug(f"Processing chunk: {chunk[:200]}...")  # Log first 200 chars
 
         try:
 
@@ -108,13 +112,13 @@ class Processor(FlowProcessor):
                     text = chunk
                 )
 
-                print("Response", defs, flush=True)
+                logger.debug(f"Definitions response: {defs}")
 
                 if type(defs) != list:
                     raise RuntimeError("Expecting array in prompt response")
 
             except Exception as e:
-                print("Prompt exception:", e, flush=True)
+                logger.error(f"Prompt exception: {e}", exc_info=True)
                 raise e
 
             triples = []
@@ -187,9 +191,9 @@ class Processor(FlowProcessor):
             )
 
         except Exception as e:
-            print("Exception: ", e, flush=True)
+            logger.error(f"Definitions extraction exception: {e}", exc_info=True)
 
-        print("Done.", flush=True)
+        logger.debug("Definitions extraction complete")
 
     @staticmethod
     def add_args(parser):

trustgraph-flow/trustgraph/extract/kg/relationships/extract.py

@@ -6,8 +6,12 @@ graph edges.
 """
 
 import json
+import logging
 import urllib.parse
 
+# Module logger
+logger = logging.getLogger(__name__)
+
 from .... schema import Chunk, Triple, Triples
 from .... schema import Metadata, Value
 from .... schema import PromptRequest, PromptResponse
@@ -78,11 +82,11 @@ class Processor(FlowProcessor):
     async def on_message(self, msg, consumer, flow):
 
         v = msg.value()
-        print(f"Indexing {v.metadata.id}...", flush=True)
+        logger.info(f"Extracting relationships from {v.metadata.id}...")
 
         chunk = v.chunk.decode("utf-8")
 
-        print(chunk, flush=True)
+        logger.debug(f"Processing chunk: {chunk[:100]}..." if len(chunk) > 100 else f"Processing chunk: {chunk}")
 
         try:
 
@@ -92,13 +96,13 @@ class Processor(FlowProcessor):
                     text = chunk
                 )
 
-                print("Response", rels, flush=True)
+                logger.debug(f"Prompt response: {rels}")
 
                 if type(rels) != list:
                     raise RuntimeError("Expecting array in prompt response")
 
             except Exception as e:
-                print("Prompt exception:", e, flush=True)
+                logger.error(f"Prompt exception: {e}", exc_info=True)
                 raise e
 
             triples = []
@@ -189,9 +193,9 @@ class Processor(FlowProcessor):
             )
 
         except Exception as e:
-            print("Exception: ", e, flush=True)
+            logger.error(f"Relationship extraction exception: {e}", exc_info=True)
 
-        print("Done.", flush=True)
+        logger.debug("Relationship extraction complete")
 
     @staticmethod
     def add_args(parser):

trustgraph-flow/trustgraph/extract/kg/topics/extract.py

@@ -6,6 +6,10 @@ get topics which are output as graph edges.
 
 import urllib.parse
 import json
+import logging
+
+# Module logger
+logger = logging.getLogger(__name__)
 
 from .... schema import Chunk, Triple, Triples, Metadata, Value
 from .... schema import chunk_ingest_queue, triples_store_queue
@@ -81,7 +85,7 @@ class Processor(ConsumerProducer):
     async def handle(self, msg):
 
         v = msg.value()
-        print(f"Indexing {v.metadata.id}...", flush=True)
+        logger.info(f"Extracting topics from {v.metadata.id}...")
 
         chunk = v.chunk.decode("utf-8")
 
@@ -110,9 +114,9 @@ class Processor(ConsumerProducer):
                 )
 
         except Exception as e:
-            print("Exception: ", e, flush=True)
+            logger.error(f"Topic extraction exception: {e}", exc_info=True)
 
-        print("Done.", flush=True)
+        logger.debug("Topic extraction complete")
 
     @staticmethod
     def add_args(parser):

trustgraph-flow/trustgraph/extract/object/row/extract.py

@@ -6,8 +6,12 @@ out a row of fields.  Output as a vector plus object.
 
 import urllib.parse
 import os
+import logging
 from pulsar.schema import JsonSchema
 
+# Module logger
+logger = logging.getLogger(__name__)
+
 from .... schema import ChunkEmbeddings, Rows, ObjectEmbeddings, Metadata
 from .... schema import RowSchema, Field
 from .... schema import chunk_embeddings_ingest_queue, rows_store_queue
@@ -75,7 +79,7 @@ class Processor(ConsumerProducer):
         flds = __class__.parse_fields(params["field"])
 
         for fld in flds:
-            print(fld)
+            logger.debug(f"Field configuration: {fld}")
 
         self.primary = None
 
@@ -142,7 +146,7 @@ class Processor(ConsumerProducer):
     async def handle(self, msg):
 
         v = msg.value()
-        print(f"Indexing {v.metadata.id}...", flush=True)
+        logger.info(f"Extracting rows from {v.metadata.id}...")
 
         chunk = v.chunk.decode("utf-8")
 
@@ -163,12 +167,12 @@ class Processor(ConsumerProducer):
                 )
 
             for row in rows:
-                print(row)
+                logger.debug(f"Extracted row: {row}")
 
         except Exception as e:
-            print("Exception: ", e, flush=True)
+            logger.error(f"Row extraction exception: {e}", exc_info=True)
 
-        print("Done.", flush=True)
+        logger.debug("Row extraction complete")
 
     @staticmethod
     def add_args(parser):