feat(telemetry): add telemetry module

Author: PeriniM

examples/openai/smart_scraper_openai.py

@@ -2,7 +2,7 @@
 Basic example of scraping pipeline using SmartScraper
 """
 
-import os
+import os, json
 from dotenv import load_dotenv
 from scrapegraphai.graphs import SmartScraperGraph
 from scrapegraphai.utils import prettify_exec_info
@@ -37,7 +37,7 @@
 )
 
 result = smart_scraper_graph.run()
-print(result)
+print(json.dumps(result, indent=4))
 
 # ************************************************
 # Get graph execution info

scrapegraphai/graphs/abstract_graph.py

@@ -26,7 +26,7 @@
     OneApi
 )
 from ..models.ernie import Ernie
-from ..utils.logging import set_verbosity_debug, set_verbosity_warning
+from ..utils.logging import set_verbosity_debug, set_verbosity_warning, set_verbosity_info
 
 from ..helpers import models_tokens
 from ..models import AzureOpenAI, Bedrock, Gemini, Groq, HuggingFace, Ollama, OpenAI, Anthropic, DeepSeek
@@ -90,7 +90,7 @@ def __init__(self, prompt: str, config: dict,
         verbose = bool(config and config.get("verbose"))
 
         if verbose:
-            set_verbosity_debug()
+            set_verbosity_info()
         else:
             set_verbosity_warning()
 

scrapegraphai/graphs/base_graph.py

@@ -1,12 +1,10 @@
-"""
-BaseGraph Module
-"""
-
 import time
 import warnings
 from langchain_community.callbacks import get_openai_callback
 from typing import Tuple
 
+# Import telemetry functions
+from ..telemetry import log_graph_execution, log_event
 
 class BaseGraph:
     """
@@ -46,12 +44,12 @@ class BaseGraph:
         ... )
     """
 
-    def __init__(self, nodes: list, edges: list, entry_point: str, use_burr: bool = False, burr_config: dict = None):
-
+    def __init__(self, nodes: list, edges: list, entry_point: str, use_burr: bool = False, burr_config: dict = None, graph_name: str = "Custom"):
         self.nodes = nodes
         self.raw_edges = edges
         self.edges = self._create_edges({e for e in edges})
         self.entry_point = entry_point.node_name
+        self.graph_name = graph_name
         self.initial_state = {}
 
         if nodes[0].node_name != entry_point.node_name:
@@ -103,12 +101,46 @@ def _execute_standard(self, initial_state: dict) -> Tuple[dict, list]:
             "total_cost_USD": 0.0,
         }
 
+        start_time = time.time()
+        error_node = None
+        source_type = None
+        llm_model = None
+        embedder_model = None
+
         while current_node_name:
             curr_time = time.time()
             current_node = next(node for node in self.nodes if node.node_name == current_node_name)
 
+            # check if there is a "source" key in the node config
+            if current_node.__class__.__name__ == "FetchNode":
+                # get the second key name of the state dictionary
+                source_type = list(state.keys())[1]
+                # quick fix for local_dir source type
+                if source_type == "local_dir":
+                    source_type = "html_dir"
+
+            # check if there is an "llm_model" variable in the class
+            if hasattr(current_node, "llm_model") and llm_model is None:
+                llm_model = current_node.llm_model
+                if hasattr(llm_model, "model_name"):
+                    llm_model = llm_model.model_name
+                elif hasattr(llm_model, "model"):
+                    llm_model = llm_model.model
+
+            # check if there is an "embedder_model" variable in the class
+            if hasattr(current_node, "embedder_model") and embedder_model is None:
+                embedder_model = current_node.embedder_model
+                if hasattr(embedder_model, "model_name"):
+                    embedder_model = embedder_model.model_name
+                elif hasattr(embedder_model, "model"):
+                    embedder_model = embedder_model.model
+
             with get_openai_callback() as cb:
-                result = current_node.execute(state)
+                try:
+                    result = current_node.execute(state)
+                except Exception as e:
+                    error_node = current_node.node_name
+                    raise e
                 node_exec_time = time.time() - curr_time
                 total_exec_time += node_exec_time
 
@@ -147,6 +179,17 @@ def _execute_standard(self, initial_state: dict) -> Tuple[dict, list]:
             "exec_time": total_exec_time,
         })
 
+        # Log the graph execution telemetry
+        graph_execution_time = time.time() - start_time
+        log_graph_execution(
+            graph_name=self.graph_name,
+            llm_model=llm_model,
+            embedder_model=embedder_model,
+            source_type=source_type,
+            execution_time=graph_execution_time,
+            error_node=error_node
+        )
+
         return state, exec_info
 
     def execute(self, initial_state: dict) -> Tuple[dict, list]:
@@ -162,7 +205,6 @@ def execute(self, initial_state: dict) -> Tuple[dict, list]:
 
         self.initial_state = initial_state
         if self.use_burr:
-
             from ..integrations import BurrBridge
 
             bridge = BurrBridge(self, self.burr_config)
@@ -190,4 +232,4 @@ def append_node(self, node):
         # add the node to the list of nodes
         self.nodes.append(node)
         # update the edges connecting the last node to the new node
-        self.edges = self._create_edges({e for e in self.raw_edges})
+        self.edges = self._create_edges({e for e in self.raw_edges})

scrapegraphai/graphs/csv_scraper_graph.py

@@ -64,7 +64,8 @@ def _create_graph(self):
                 (fetch_node, rag_node),
                 (rag_node, generate_answer_node)
             ],
-            entry_point=fetch_node
+            entry_point=fetch_node,
+            graph_name=self.__class__.__name__
         )
 
     def run(self) -> str:

scrapegraphai/graphs/csv_scraper_multi_graph.py

@@ -100,7 +100,8 @@ def _create_graph(self) -> BaseGraph:
             edges=[
                 (graph_iterator_node, merge_answers_node),
             ],
-            entry_point=graph_iterator_node
+            entry_point=graph_iterator_node,
+            graph_name=self.__class__.__name__
         )
 
     def run(self) -> str:

scrapegraphai/graphs/deep_scraper_graph.py

@@ -141,7 +141,8 @@ def _create_repeated_graph(self) -> BaseGraph:
                 (search_node, graph_iterator_node),
                 (graph_iterator_node, merge_answers_node)
             ],
-            entry_point=fetch_node
+            entry_point=fetch_node,
+            graph_name=self.__class__.__name__
         )
 
 

scrapegraphai/graphs/json_scraper_graph.py

@@ -89,7 +89,8 @@ def _create_graph(self) -> BaseGraph:
                 (fetch_node, rag_node),
                 (rag_node, generate_answer_node)
             ],
-            entry_point=fetch_node
+            entry_point=fetch_node,
+            graph_name=self.__class__.__name__
         )
 
     def run(self) -> str:

scrapegraphai/graphs/json_scraper_multi_graph.py

@@ -104,7 +104,8 @@ def _create_graph(self) -> BaseGraph:
             edges=[
                 (graph_iterator_node, merge_answers_node),
             ],
-            entry_point=graph_iterator_node
+            entry_point=graph_iterator_node,
+            graph_name=self.__class__.__name__
         )
 
     def run(self) -> str:

scrapegraphai/graphs/omni_scraper_graph.py

@@ -122,7 +122,8 @@ def _create_graph(self) -> BaseGraph:
                 (image_to_text_node, rag_node),
                 (rag_node, generate_answer_omni_node)
             ],
-            entry_point=fetch_node
+            entry_point=fetch_node,
+            graph_name=self.__class__.__name__
         )
 
     def run(self) -> str:

scrapegraphai/graphs/omni_search_graph.py

@@ -115,7 +115,8 @@ def _create_graph(self) -> BaseGraph:
                 (search_internet_node, graph_iterator_node),
                 (graph_iterator_node, merge_answers_node)
             ],
-            entry_point=search_internet_node
+            entry_point=search_internet_node,
+            graph_name=self.__class__.__name__
         )
 
     def run(self) -> str: