Repository: deepset-ai/haystack
Pull request: fix: Remove pipeline serialization from telemetry code
### Related Issues

- fixes #7255
~~probably also fixing #7277 we need to double check ~~
- fixes #7277 

### Proposed Changes:

Instead of using pipeline serialization to get all components, we now loop through the nodes in the pipeline graph.

### How did you test it?

<!-- unit tests, integration tests, manual verification, instructions for manual tests -->

### Notes for the reviewer

<!-- E.g. point out section where the reviewer  -->

### Checklist

- I have read the [contributors guidelines](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md) and the [code of conduct](https://github.com/deepset-ai/haystack/blob/main/code_of_conduct.txt)
- I have updated the related issue with new insights and changes
- I added unit tests and updated the docstrings
- I've used one of the [conventional commit types](https://www.conventionalcommits.org/en/v1.0.0/) for my PR title: `fix:`, `feat:`, `build:`, `chore:`, `ci:`, `docs:`, `style:`, `refactor:`, `perf:`, `test:`.
- I documented my code
- I ran [pre-commit hooks](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md#installation) and fixed any issue


Problem statement:
ValueError is raised by telemetry code in `component.to_dict()` if a TokenSecret is used
A ValueError is raised by telemetry code in `component.to_dict()` if a TokenSecret is used in any component of a pipeline.

In telemetry code, we currently serialize the whole pipeline to then loop through the components and add their name and any additional telemetry data to a telemetry event.
We should not serialize the whole pipeline to do that. TokenSecrets cannot be serialized and raise a ValueError.

Problem when using pipeline serialization
Hello,

I woul like to use pipeline serialization. Below is my code for the llm

  ```
llm:
    init_parameters:
      huggingface_pipeline_kwargs:
        model: mistralai/Mixtral-8x7B-Instruct-v0.1
        device_map: auto
        model_kwargs:
          bnb_4bit_compute_dtype: torch.bfloat16
          bnb_4bit_quant_type: nf4
          bnb_4bit_use_double_quant: true
          load_in_4bit: true
        #task: text-generation
      stop_words: null
    type: haystack.components.generators.hugging_face_local.HuggingFaceLocalGenerator
    
```
this code will generate the below error:

~/.local/lib/python3.9/site-packages/bitsandbytes/nn/modules.py in forward(self, x)
    251         inp_dtype = x.dtype
    252         if self.compute_dtype is not None:
--> 253             x = x.to(self.compute_dtype)
    254 
    255         bias = None if self.bias is None else self.bias.to(self.compute_dtype)

RuntimeError: Invalid device string: 'torch.bfloat16'


I dont know if there is another way to pass the compute_type 

Best regards

Hints from issue discussion:
Hey @Asma-droid...

Try with "torch.bfloat16" (quotes included).
hello @anakin87 . I still have the same problem
Please provide more information.
Which specific beta version are you using?
How did you generate such serialized pipeline?
How are you deserializing that pipeline?

With this information, tomorrow I will have a better look...


@anakin87  Thanks a lot for your reactivity!

I'am using haystack-ai  2.0.0b8.

```
my pipeline is:
query_pipeline = Pipeline()
query_pipeline.add_component("text_embedder", SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"))
query_pipeline.add_component("retriever", ElasticsearchEmbeddingRetriever(document_store=document_store))
query_pipeline.add_component("prompt_builder", prompt_builder)
query_pipeline.add_component("llm", llm)
query_pipeline.add_component("AnswerBuilder", AnswerBuilder())



query_pipeline.connect("text_embedder.embedding", "retriever.query_embedding")
query_pipeline.connect("retriever", "prompt_builder.documents")
query_pipeline.connect("prompt_builder", "llm")
query_pipeline.connect("llm.replies", "AnswerBuilder")
#query_pipeline.connect("llm.meta", "AnswerBuilder")
query_pipeline.connect("retriever", "AnswerBuilder.documents")
```

i generated automatically the serialized pipeline using query_pipeline.dumps(). i have got

```
components:
  AnswerBuilder:
    init_parameters:
      pattern: null
      reference_pattern: null
    type: haystack.components.builders.answer_builder.AnswerBuilder
  llm:
    init_parameters:
      generation_kwargs:
        max_new_tokens: 350
        return_full_text: false
      huggingface_pipeline_kwargs:
        device_map: auto
        model: mistralai/Mixtral-8x7B-Instruct-v0.1
        model_kwargs:
          bnb_4bit_compute_dtype: torch.bfloat16
          bnb_4bit_quant_type: nf4
          bnb_4bit_use_double_quant: true
          device_map: auto
          load_in_4bit: true
        task: text-generation
      stop_words: null
      token:
        env_vars:
        - HF_API_TOKEN
        strict: false
        type: env_var
    type: haystack.components.generators.hugging_face_local.HuggingFaceLocalGenerator
  prompt_builder:
    init_parameters:
      template: "\nAnswer the query based on the provided context. Please if the query\
        \ is in french, provide the answer in french also \nIf the context does not\
        \ contain the answer, say 'Answer not found'.\nContext:\n{% for doc in documents\
        \ %}\n  {{ doc.content }}\n{% endfor %}\nquery: {{query}}\nAnswer:\n"
    type: haystack.components.builders.prompt_builder.PromptBuilder
  retriever:
    init_parameters:
      document_store:
        init_parameters:
          embedding_similarity_function: cosine
          hosts: http://127.0.0.1:9200
          index: document-cfdt-new-asma
        type: haystack_integrations.document_stores.elasticsearch.document_store.ElasticsearchDocumentStore
      filters: {}
      num_candidates: null
      top_k: 10
    type: haystack_integrations.components.retrievers.elasticsearch.embedding_retriever.ElasticsearchEmbeddingRetriever
  text_embedder:
    init_parameters:
      batch_size: 32
      device:
        device: cuda:0
        type: single
      model: sentence-transformers/all-MiniLM-L6-v2
      normalize_embeddings: false
      prefix: ''
      progress_bar: true
      suffix: ''
      token:
        env_vars:
        - HF_API_TOKEN
        strict: false
        type: env_var
    type: haystack.components.embedders.sentence_transformers_text_embedder.SentenceTransformersTextEmbedder
connections:
- receiver: retriever.query_embedding
  sender: text_embedder.embedding
- receiver: prompt_builder.documents
  sender: retriever.documents
- receiver: AnswerBuilder.documents
  sender: retriever.documents
- receiver: llm.prompt
  sender: prompt_builder.prompt
- receiver: AnswerBuilder.replies
  sender: llm.replies
max_loops_allowed: 100
metadata: {}
```

I have saved my pipeline in a file:

and i have executed this code 

```
from haystack import Pipeline
with open("./src/pipelines/rag_pipeline.yaml", "rb") as f:
    rag_piepeline = Pipeline.loads(f)
```

i have also traied Pipeline.loads(f)
   
i got the problem when i launch

```
query="What is Haystack?"
result = rag_piepeline.run(
     {"text_embedder": {"text": query},
                     "retriever": {"top_k": 10},
                     "prompt_builder": {"query": query},
                     "llm":{"generation_kwargs": {"max_new_tokens": 1000}},
                     "AnswerBuilder": {"query": query}})
```
Please also provide the code to initialize the llm.
@anakin87 

```
llm:
  init_parameters:
    huggingface_pipeline_kwargs:
      model: mistralai/Mixtral-8x7B-Instruct-v0.1
      device_map: auto
      model_kwargs:
        bnb_4bit_compute_dtype: torch.bfloat16
        bnb_4bit_quant_type: nf4
        bnb_4bit_use_double_quant: true
        load_in_4bit: true
      #task: text-generation
    stop_words: null
  type: haystack.components.generators.hugging_face_local.HuggingFaceLocalGenerator
```

and
```

llm = HuggingFaceLocalGenerator(model="mistralai/Mixtral-8x7B-Instruct-v0.1",
                                          task="text-generation",
                                          huggingface_pipeline_kwargs={"device_map":"auto",
                                                   "model_kwargs":{"load_in_4bit":True,
                                                                   "bnb_4bit_use_double_quant":True,
                                                                   "bnb_4bit_quant_type":"nf4",
                                                                   "bnb_4bit_compute_dtype":torch.bfloat16}})
```
I have made some experiments and it seems to be a bug. I will add some information.

**Using the generator on its own works well**
```python
from haystack.components.generators import HuggingFaceLocalGenerator
from haystack import Pipeline
import torch

pipe = Pipeline()

prompt_template = "Answer the following question: {{question}}"
prompt_builder = PromptBuilder(prompt_template)

llm = HuggingFaceLocalGenerator(model="microsoft/phi-2",
                                           huggingface_pipeline_kwargs={"device_map":"auto",
                                                   "model_kwargs":{"load_in_4bit":True,
                                                                   "bnb_4bit_use_double_quant":True,
                                                                   "bnb_4bit_quant_type":"nf4",
                                                                   "bnb_4bit_compute_dtype":torch.bfloat16})

llm.warm_up()

llm.run("What is the capital of Germany? Answer:")
```

**Using the generator in a Pipeline causes errors**
(probably related to serialization)
```python
from haystack.components.generators import HuggingFaceLocalGenerator
from haystack import Pipeline
from haystack.components.builders import PromptBuilder
import torch

pipe = Pipeline()

prompt_template = "Answer the following question: {{question}}"
prompt_builder = PromptBuilder(prompt_template)

llm = HuggingFaceLocalGenerator(
    model="microsoft/phi-2",
    huggingface_pipeline_kwargs={
        "device_map": "auto",
        "model_kwargs": {
            "load_in_4bit": True,
            "bnb_4bit_use_double_quant": True,
            "bnb_4bit_quant_type": "nf4",
            "bnb_4bit_compute_dtype": torch.bfloat16,
        },
    },
)

# llm.warm_up()
# llm.run("What is the capital of Germany? Answer:")

pipe.add_component("prompt_builder", prompt_builder)
pipe.add_component("llm", llm)
pipe.connect("prompt_builder", "llm")

print(pipe.run({"question": "What is the capital of Germany?"}))

```

`AttributeError: module 'torch' has no attribute 'torch.bfloat16'`

Exactely, i have face the same errors.  As i'am using serialized pipeline, i dont know how to solve the issue
I have verified that this is the same problem expressed in #7255 and it is related to incorrect serialization in telemetry.
We will solve this soon.

In the meantime, you can disable telemetry:
`export HAYSTACK_TELEMETRY_ENABLED="False"`

Commit message: add test for pipeline.walk() with cycles

Patch to review:
diff --git a/haystack/core/pipeline/pipeline.py b/haystack/core/pipeline/pipeline.py
index 0d70b5be0b..7a713daf40 100644
--- a/haystack/core/pipeline/pipeline.py
+++ b/haystack/core/pipeline/pipeline.py
@@ -7,7 +7,7 @@
 from copy import copy, deepcopy
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Dict, List, Mapping, Optional, Set, TextIO, Tuple, Type, TypeVar, Union
+from typing import Any, Dict, Iterator, List, Mapping, Optional, Set, TextIO, Tuple, Type, TypeVar, Union
 
 import networkx  # type:ignore
 
@@ -558,6 +558,16 @@ def draw(self, path: Path) -> None:
         image_data = _to_mermaid_image(self.graph)
         Path(path).write_bytes(image_data)
 
+    def walk(self) -> Iterator[Tuple[str, Component]]:
+        """
+        Walks the pipeline, yielding tuples of name and instance of each component exactly once in no guaranteed order.
+
+        :returns:
+            An iterator of tuples of component name and component instance.
+        """
+        for component_name, instance in self.graph.nodes(data="instance"):
+            yield component_name, instance
+
     def warm_up(self):
         """
         Make sure all nodes are warm.
diff --git a/haystack/core/serialization.py b/haystack/core/serialization.py
index e2f4ba1f78..8ed7fe5dfe 100644
--- a/haystack/core/serialization.py
+++ b/haystack/core/serialization.py
@@ -47,6 +47,18 @@ def component_to_dict(obj: Any) -> Dict[str, Any]:
     return default_to_dict(obj, **init_parameters)
 
 
+def generate_qualified_class_name(cls: Type[object]) -> str:
+    """
+    Generates a qualified class name for a class.
+
+    :param cls:
+        The class whose qualified name is to be generated.
+    :returns:
+        The qualified name of the class.
+    """
+    return f"{cls.__module__}.{cls.__name__}"
+
+
 def component_from_dict(cls: Type[object], data: Dict[str, Any]) -> Any:
     """
     Creates a component instance from a dictionary. If a `from_dict` method is present in the
@@ -103,7 +115,7 @@ def to_dict(self):
     :returns:
         A dictionary representation of the instance.
     """
-    return {"type": f"{obj.__class__.__module__}.{obj.__class__.__name__}", "init_parameters": init_parameters}
+    return {"type": generate_qualified_class_name(type(obj)), "init_parameters": init_parameters}
 
 
 def default_from_dict(cls: Type[object], data: Dict[str, Any]) -> Any:
@@ -130,6 +142,6 @@ def default_from_dict(cls: Type[object], data: Dict[str, Any]) -> Any:
     init_params = data.get("init_parameters", {})
     if "type" not in data:
         raise DeserializationError("Missing 'type' in serialization data")
-    if data["type"] != f"{cls.__module__}.{cls.__name__}":
+    if data["type"] != generate_qualified_class_name(cls):
         raise DeserializationError(f"Class '{data['type']}' can't be deserialized as '{cls.__name__}'")
     return cls(**init_params)
diff --git a/haystack/telemetry/_telemetry.py b/haystack/telemetry/_telemetry.py
index cb41cfac15..13f3386368 100644
--- a/haystack/telemetry/_telemetry.py
+++ b/haystack/telemetry/_telemetry.py
@@ -10,6 +10,7 @@
 import yaml
 
 from haystack import logging as haystack_logging
+from haystack.core.serialization import generate_qualified_class_name
 from haystack.telemetry._environment import collect_system_specs
 
 if TYPE_CHECKING:
@@ -144,18 +145,17 @@ def pipeline_running(pipeline: "Pipeline") -> Optional[Tuple[str, Dict[str, Any]
     pipeline._last_telemetry_sent = datetime.datetime.now()
 
     # Collect info about components
-    pipeline_description = pipeline.to_dict()
     components: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
-    for component_name, component in pipeline_description["components"].items():
-        instance = pipeline.get_component(component_name)
+    for component_name, instance in pipeline.walk():
+        component_qualified_class_name = generate_qualified_class_name(type(instance))
         if hasattr(instance, "_get_telemetry_data"):
             telemetry_data = getattr(instance, "_get_telemetry_data")()
             try:
-                components[component["type"]].append({"name": component_name, **telemetry_data})
+                components[component_qualified_class_name].append({"name": component_name, **telemetry_data})
             except TypeError:
-                components[component["type"]].append({"name": component_name})
+                components[component_qualified_class_name].append({"name": component_name})
         else:
-            components[component["type"]].append({"name": component_name})
+            components[component_qualified_class_name].append({"name": component_name})
 
     # Data sent to Posthog
     return "Pipeline run (2.x)", {
diff --git a/releasenotes/notes/remove-serialization-from-telemetry-a054f7f26f277d2b.yaml b/releasenotes/notes/remove-serialization-from-telemetry-a054f7f26f277d2b.yaml
new file mode 100644
index 0000000000..82962d877d
--- /dev/null
+++ b/releasenotes/notes/remove-serialization-from-telemetry-a054f7f26f277d2b.yaml
@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+    Fix telemetry code that could cause a ValueError by trying to serialize a pipeline.
+    Telemetry code does not serialize pipelines anymore.
diff --git a/test/core/pipeline/test_pipeline.py b/test/core/pipeline/test_pipeline.py
index 48096276cb..c11f7a53f8 100644
--- a/test/core/pipeline/test_pipeline.py
+++ b/test/core/pipeline/test_pipeline.py
@@ -659,3 +659,73 @@ def test_describe_no_outputs():
 def test_from_template():
     pipe = Pipeline.from_template(PredefinedPipeline.INDEXING)
     assert pipe.get_component("cleaner")
+
+
+def test_walk_pipeline_with_no_cycles():
+    """
+    This pipeline has two source nodes, source1 and source2, one hello3 node in between, and one sink node, joiner.
+    pipeline.walk() should return each component exactly once. The order is not guaranteed.
+    """
+
+    @component
+    class Hello:
+        @component.output_types(output=str)
+        def run(self, word: str):
+            """
+            Takes a string in input and returns "Hello, <string>!" in output.
+            """
+            return {"output": f"Hello, {word}!"}
+
+    @component
+    class Joiner:
+        @component.output_types(output=str)
+        def run(self, word1: str, word2: str):
+            """
+            Takes two strings in input and returns "Hello, <string1> and <string2>!" in output.
+            """
+            return {"output": f"Hello, {word1} and {word2}!"}
+
+    pipeline = Pipeline()
+    source1 = Hello()
+    source2 = Hello()
+    hello3 = Hello()
+    joiner = Joiner()
+    pipeline.add_component("source1", source1)
+    pipeline.add_component("source2", source2)
+    pipeline.add_component("hello3", hello3)
+    pipeline.add_component("joiner", joiner)
+
+    pipeline.connect("source1", "joiner.word1")
+    pipeline.connect("source2", "hello3")
+    pipeline.connect("hello3", "joiner.word2")
+
+    expected_components = [("source1", source1), ("source2", source2), ("joiner", joiner), ("hello3", hello3)]
+    assert sorted(expected_components) == sorted(pipeline.walk())
+
+
+def test_walk_pipeline_with_cycles():
+    """
+    This pipeline consists of one component, which would run three times in a loop.
+    pipeline.walk() should return this component exactly once. The order is not guaranteed.
+    """
+
+    @component
+    class Hello:
+        def __init__(self):
+            self.iteration_counter = 0
+
+        @component.output_types(intermediate=str, final=str)
+        def run(self, word: str, intermediate: Optional[str] = None):
+            """
+            Takes a string in input and returns "Hello, <string>!" in output.
+            """
+            if self.iteration_counter < 3:
+                self.iteration_counter += 1
+                return {"intermediate": f"Hello, {intermediate or word}!"}
+            return {"final": f"Hello, {intermediate or word}!"}
+
+    pipeline = Pipeline()
+    hello = Hello()
+    pipeline.add_component("hello", hello)
+    pipeline.connect("hello.intermediate", "hello.intermediate")
+    assert [("hello", hello)] == list(pipeline.walk())
diff --git a/test/core/test_serialization.py b/test/core/test_serialization.py
index 1534ceef0c..aba90ce75f 100644
--- a/test/core/test_serialization.py
+++ b/test/core/test_serialization.py
@@ -7,7 +7,7 @@
 from haystack.core.component import component
 from haystack.core.errors import DeserializationError
 from haystack.testing import factory
-from haystack.core.serialization import default_to_dict, default_from_dict
+from haystack.core.serialization import default_to_dict, default_from_dict, generate_qualified_class_name
 
 
 def test_default_component_to_dict():
@@ -77,3 +77,10 @@ def test_from_dict_import_type():
     from haystack.testing.sample_components.greet import Greet
 
     assert type(p.get_component("greeter")) == Greet
+
+
+def test_get_qualified_class_name():
+    MyComponent = factory.component_class("MyComponent")
+    comp = MyComponent()
+    res = generate_qualified_class_name(type(comp))
+    assert res == "haystack.testing.factory.MyComponent"
diff --git a/test/test_telemetry.py b/test/test_telemetry.py
index 7e12ad6459..d83403c6bd 100644
--- a/test/test_telemetry.py
+++ b/test/test_telemetry.py
@@ -1,9 +1,9 @@
 import datetime
 from unittest.mock import Mock, patch
-import pytest
 
 from haystack import Pipeline, component
 from haystack.telemetry._telemetry import pipeline_running
+from haystack.utils.auth import Secret, TokenSecret
 
 
 @patch("haystack.telemetry._telemetry.telemetry")
@@ -51,3 +51,32 @@ def run(self):
             "components": {"test.test_telemetry.Component": [{"name": "component", "key": "values"}]},
         },
     )
+
+
+@patch("haystack.telemetry._telemetry.telemetry")
+def test_pipeline_running_with_non_serializable_component(telemetry):
+    telemetry.send_event = Mock()
+
+    @component
+    class Component:
+        def __init__(self, api_key: Secret = TokenSecret("api_key")):
+            self.api_key = api_key
+
+        def _get_telemetry_data(self):
+            return {"key": "values"}
+
+        @component.output_types(value=int)
+        def run(self):
+            pass
+
+    pipe = Pipeline()
+    pipe.add_component("component", Component())
+    pipeline_running(pipe)
+    telemetry.send_event.assert_called_once_with(
+        "Pipeline run (2.x)",
+        {
+            "pipeline_id": str(id(pipe)),
+            "runs": 1,
+            "components": {"test.test_telemetry.Component": [{"name": "component", "key": "values"}]},
+        },
+    )


Code review comments:
diff_hunk: @@ -144,18 +145,17 @@ def pipeline_running(pipeline: "Pipeline") -> Optional[Tuple[str, Dict[str, Any]
     pipeline._last_telemetry_sent = datetime.datetime.now()
 
     # Collect info about components
-    pipeline_description = pipeline.to_dict()
     components: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
-    for component_name, component in pipeline_description["components"].items():
-        instance = pipeline.get_component(component_name)
+    for component_name, instance in pipeline.walk():
+        component_qualified_class_name = generate_qualified_class_name(type(instance))
         if hasattr(instance, "_get_telemetry_data"):
             telemetry_data = getattr(instance, "_get_telemetry_data")()
             try:
-                components[component["type"]].append({"name": component_name, **telemetry_data})
+                components[component_qualified_class_name].append({"name": component_name, **telemetry_data})
             except TypeError:
-                components[component["type"]].append({"name": component_name})
+                components[component_qualified_class_name].append({"name": component_name})
path: haystack/telemetry/_telemetry.py
text: @user1:
Why do we need exception handling here? Looks like we expect the method to return a dict, so let's just check the type of the return value, raise an error if it's not a Dict and then unpack it.

@author:
We have the guiding principle that telemetry should never let the execution of the user's pipeline fail. That's why we use a wrapper that catches any error raised by telemetry and sends a debug log message instead: https://github.com/deepset-ai/haystack/blob/312ba9ebf038cd3d04aaa2bf77e00dc1c8c10e9a/haystack/telemetry/_telemetry.py#L116

I changed the code here so that it raises a TypeError now. This error will be caught and become a log message then and no telemetry event will be sent. With the previous version of the code we sent an event but just without the telemetry_data.