feat: add component name and type to StreamingChunk

haystack/components/generators/chat/openai.py

@@ -574,10 +574,21 @@ def _convert_chat_completion_chunk_to_streaming_chunk(self, chunk: ChatCompletio
         if len(chunk.choices) == 0:
             return StreamingChunk(content="", meta={"model": chunk.model, "received_at": datetime.now().isoformat()})
 
+        # get the name assigned to component in the pipeline
+        component_name = str(self.__component_name__) if getattr(self, "__component_name__", None) is not None else None
+
         # we stream the content of the chunk if it's not a tool or function call
         choice: ChunkChoice = chunk.choices[0]
         content = choice.delta.content or ""
         chunk_message = StreamingChunk(content)
+
+        # add the component info to the chunk
+        # component_name will be available if the component is added to a pipeline
+        component_info = {"component_type": f"{self.__class__.__module__}.{self.__class__.__name__}"}
+        if component_name:
+            component_info["component_name"] = component_name
+        chunk_message.component_info.update(component_info)
+
         # but save the tool calls and function call in the meta if they are present
         # and then connect the chunks in the _convert_streaming_chunks_to_chat_message method
         chunk_message.meta.update(

haystack/core/pipeline/base.py

@@ -360,6 +360,7 @@ def add_component(self, name: str, instance: Component) -> None:
             raise PipelineError(msg)
 
         setattr(instance, "__haystack_added_to_pipeline__", self)
+        setattr(instance, "__component_name__", name)
 
         # Add component to the graph, disconnected
         logger.debug("Adding component '{component_name}' ({component})", component_name=name, component=instance)

haystack/dataclasses/streaming_chunk.py

@@ -17,10 +17,13 @@ class StreamingChunk:
 
     :param content: The content of the message chunk as a string.
     :param meta: A dictionary containing metadata related to the message chunk.
+    :param component_info: A dictionary containing information about the component that generated the chunk,
+        such as the component name and type.
     """
 
     content: str
     meta: Dict[str, Any] = field(default_factory=dict, hash=False)
+    component_info: Dict[str, Any] = field(default_factory=dict, hash=False)
 
 
 SyncStreamingCallbackT = Callable[[StreamingChunk], None]