diff --git a/trustgraph-flow/trustgraph/extract/kg/agent/extract.py b/trustgraph-flow/trustgraph/extract/kg/agent/extract.py index 9b15b44c..59fec208 100644 --- a/trustgraph-flow/trustgraph/extract/kg/agent/extract.py +++ b/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) - print("Configuration reload failed", flush=True) + logger.error(f"Configuration reload exception: {e}", exc_info=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): diff --git a/trustgraph-flow/trustgraph/extract/kg/definitions/extract.py b/trustgraph-flow/trustgraph/extract/kg/definitions/extract.py index 66571478..1d414b7e 100755 --- a/trustgraph-flow/trustgraph/extract/kg/definitions/extract.py +++ b/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): diff --git a/trustgraph-flow/trustgraph/extract/kg/relationships/extract.py b/trustgraph-flow/trustgraph/extract/kg/relationships/extract.py index dafee77d..6d461997 100755 --- a/trustgraph-flow/trustgraph/extract/kg/relationships/extract.py +++ b/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): diff --git a/trustgraph-flow/trustgraph/extract/kg/topics/extract.py b/trustgraph-flow/trustgraph/extract/kg/topics/extract.py index 84ab6681..129cc64c 100755 --- a/trustgraph-flow/trustgraph/extract/kg/topics/extract.py +++ b/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): diff --git a/trustgraph-flow/trustgraph/extract/object/row/extract.py b/trustgraph-flow/trustgraph/extract/object/row/extract.py index 9ccf3370..e262c1cb 100755 --- a/trustgraph-flow/trustgraph/extract/object/row/extract.py +++ b/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):