quiet logger

r2r/core/logging/kv_logger.py

@@ -64,9 +64,6 @@ def __init__(self, config: LoggingConfig):
         self.logging_path = config.logging_path or os.getenv(
             "LOCAL_DB_PATH", "local.sqlite"
         )
-        logger.info(
-            f"Initializing LocalKVLoggingProvider with config: {config}"
-        )
         if not self.logging_path:
             raise ValueError(
                 "Please set the environment variable LOCAL_DB_PATH."
@@ -219,9 +216,6 @@ def __init__(self, config: PostgresLoggingConfig):
         self.log_info_table = config.log_info_table
         self.config = config
         self.conn = None
-        logger.info(
-            f"Initializing PostgresKVLoggingProvider with config: {config}"
-        )
         if not os.getenv("POSTGRES_DBNAME"):
             raise ValueError(
                 "Please set the environment variable POSTGRES_DBNAME."
@@ -509,6 +503,10 @@ def get_instance(cls):
     def configure(
         cls, logging_config: Optional[LoggingConfig] = LoggingConfig()
     ):
+        logger.info(
+            f"Initializing KVLoggingSingleton with config: {logging_config}"
+        )
+
         if not cls._is_configured:
             cls._config = logging_config
             cls._is_configured = True
@@ -526,8 +524,9 @@ async def log(
         try:
             async with cls.get_instance() as provider:
                 await provider.log(log_id, key, value, is_info_log=is_info_log)
+
         except Exception as e:
-            logger.error(f"Error logging data: {e}")
+            logger.error(f"Error logging data {(log_id, key, value)}: {e}")
 
     @classmethod
     async def get_run_info(

r2r/core/pipeline/base_pipeline.py

@@ -30,11 +30,13 @@ def __init__(
         self,
         pipe_logger: Optional[KVLoggingSingleton] = None,
         run_manager: Optional[RunManager] = None,
+        log_run_info: bool = True,
     ):
         self.pipes: list[AsyncPipe] = []
         self.upstream_outputs: list[list[dict[str, str]]] = []
         self.pipe_logger = pipe_logger or KVLoggingSingleton()
         self.run_manager = run_manager or RunManager(self.pipe_logger)
+        self.log_run_info = log_run_info
         self.futures = {}
         self.level = 0
 
@@ -73,11 +75,12 @@ async def run(
         self.state = state or AsyncState()
         current_input = input
         async with manage_run(run_manager, self.pipeline_type):
-            await run_manager.log_run_info(
-                key="pipeline_type",
-                value=self.pipeline_type,
-                is_info_log=True,
-            )
+            if self.log_run_info:
+                await run_manager.log_run_info(
+                    key="pipeline_type",
+                    value=self.pipeline_type,
+                    is_info_log=True,
+                )
             try:
                 for pipe_num in range(len(self.pipes)):
                     config_name = self.pipes[pipe_num].config.name
@@ -345,13 +348,13 @@ def add_pipe(
             self.parsing_pipe = pipe
         elif kg_pipe:
             if not self.kg_pipeline:
-                self.kg_pipeline = Pipeline()
+                self.kg_pipeline = Pipeline(log_run_info=False)
             self.kg_pipeline.add_pipe(
                 pipe, add_upstream_outputs, *args, **kwargs
             )
         elif embedding_pipe:
             if not self.embedding_pipeline:
-                self.embedding_pipeline = Pipeline()
+                self.embedding_pipeline = Pipeline(log_run_info=False)
             self.embedding_pipeline.add_pipe(
                 pipe, add_upstream_outputs, *args, **kwargs
             )