Repository: deepset-ai/haystack
Pull request: feat: Add completion start time timestamp to relevant generators
### Why:
Enhances Langfuse integration to track the time-to-first-token (TTFT) in streaming responses. The new entry in chat message metadata `completion_start_time` enables TTFT tracking in Langfuse. We'll separately update all other relevant generators in https://github.com/deepset-ai/haystack-core-integrations/ 

- fixes https://github.com/deepset-ai/haystack-core-integrations/issues/1261

### What:
- Added `completion_start_time` metadata in the `Hugging Face API` and `OpenAI` chat and non-chat generators to record the TTFT.
- Modified test cases to validate the inclusion of the `completion_start_time` and its format correctness.

### How can it be used:
This change allows developers to track the exact start time of streaming completions, aiding in performance diagnostics

### How did you test it:
Included updated unit tests for both `Hugging Face API` and `OpenAI` chat and non-chat generators to ensure the presence and correct ISO format of `completion_start_time` metadata. These tests verify that the timestamps are recorded and formatted correctly during streaming responses. 

### Notes for the reviewer:
Ensure close attention to the new metadata handling and its correctness across different components. Testing has validated the format, but real integration checks are recommended for full assurance.


Problem statement:
Langfuse integration: completionStartedAt is not logged, required for streaming metrics like TTFT and TPM/TPS
**Describe the bug**

When using the Langfuse integration, time to first token metrics are not available (the report I received mentioned Azure OpenAI but this might apply to all LLMs).

Langfuse supports time to first token via the “completionStartedAt” attribute on generations.

**To Reproduce**

Use Langfuse integration with AzureOpenAI in Haystack.

**Describe your environment (please complete the following information):**

Reported by user. Can try to get details on this if helpful.

Hints from issue discussion:
This is a bit vague as it was reported by a user but I assume the attribute is not set at all and thus it could be sufficiently detailed.

I think this would be a great addition to the integration.
Hey @marcklingen we currently [use](https://github.com/deepset-ai/haystack-core-integrations/blob/main/integrations/langfuse/src/haystack_integrations/tracing/langfuse/tracer.py#L204) `completion_start_time` as indicated in Generation section of your [docs](https://langfuse.com/docs/sdk/python/low-level-sdk) here.

Has this changed? IIRC Hassieb mentioned some upcoming changes in the API...
@marcklingen @hassiebp it seems that previously the "completion_start_time" key/value pair was automatically injected in meta of chat messages - by langfuse. I don't see this happening any more (using your latest python sdk). Look I'm not 100% sure, with so many moving parts everywhere but here is what I tried. When I set `completion_start_time` value manually into chat message myself our current integration [code](https://github.com/deepset-ai/haystack-core-integrations/blob/main/integrations/langfuse/src/haystack_integrations/tracing/langfuse/tracer.py#L204) picks it up and everything is displayed in the web tracer correctly. Any advice or pointers appreciated. 
Hi @vblagoje - the completion_start_time is not available on a chat message meta but only on the returned generator _after_ the first chunk has been consumed:

```python
from langfuse.openai import openai

completion = openai.chat.completions.create(
    stream=True,
    model="gpt-4o-2024-11-20",
    messages=[
        {"role": "system", "content": "You are a very comedian."},
        {"role": "user", "content": "Give me a one-liner"},
    ],
)

print(completion.completion_start_time)  # prints None

for chunk in completion:
    print(chunk.choices[0].delta.content or "", end="")


print("\n" + str(completion.completion_start_time))  # prints datetime

openai.flush_langfuse()
```

Let me know if I should expand further on anything! 🙏🏾 
Aha, but we don't use your instrument openai library but a pure openai library. And then we hook our internal tracer into Langfuse as you can see [here](https://github.com/deepset-ai/haystack-core-integrations/blob/main/integrations/langfuse/src/haystack_integrations/tracing/langfuse/tracer.py).

Any ideas how we can get a hold of this data there?
@vblagoje I'm unfortunately not very familiar with how the Haystack integration is built. Something along these lines should work though, i.e. capturing timestamp on each generated chunk and storing it in the generation `meta`, which is then used in the Langfuse Tracer to update the `completion_start_time`. 

(example Anthropic)

![Image](https://github.com/user-attachments/assets/680f2a4a-7950-47e0-b1de-b3aa9eab23b7)
Aha ok, thanks a lot @hassiebp that's what I thought as well. I already made a local branch with our OpenAIChatGenerator that stores this data and then everything gets properly calculated and displayed in langfuse webui. 🙏

Commit message: Add completion_start_time timestamping to non-chat generators

Patch to review:
diff --git a/haystack/components/generators/chat/hugging_face_api.py b/haystack/components/generators/chat/hugging_face_api.py
index 50a730a01f..97d7f54a97 100644
--- a/haystack/components/generators/chat/hugging_face_api.py
+++ b/haystack/components/generators/chat/hugging_face_api.py
@@ -2,6 +2,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+from datetime import datetime
 from typing import Any, Callable, Dict, Iterable, List, Optional, Union
 
 from haystack import component, default_from_dict, default_to_dict, logging
@@ -259,6 +260,8 @@ def _run_streaming(self, messages: List[Dict[str, str]], generation_kwargs: Dict
         )
 
         generated_text = ""
+        chunks_meta = []
+        first_chunk_time = None
 
         for chunk in api_output:
             # n is unused, so the API always returns only one choice
@@ -272,9 +275,14 @@ def _run_streaming(self, messages: List[Dict[str, str]], generation_kwargs: Dict
 
             finish_reason = choice.finish_reason
 
-            meta = {}
-            if finish_reason:
-                meta["finish_reason"] = finish_reason
+            meta = {
+                "finish_reason": finish_reason if finish_reason else None,
+                "received_at": datetime.now().isoformat(),
+            }
+            chunks_meta.append(meta)
+
+            if first_chunk_time is None:
+                first_chunk_time = meta["received_at"]
 
             stream_chunk = StreamingChunk(text, meta)
             self.streaming_callback(stream_chunk)  # type: ignore # streaming_callback is not None (verified in the run method)
@@ -285,6 +293,7 @@ def _run_streaming(self, messages: List[Dict[str, str]], generation_kwargs: Dict
                 "finish_reason": finish_reason,
                 "index": 0,
                 "usage": {"prompt_tokens": 0, "completion_tokens": 0},  # not available in streaming
+                "completion_start_time": first_chunk_time,
             }
         )
 
diff --git a/haystack/components/generators/chat/openai.py b/haystack/components/generators/chat/openai.py
index 0b699e3bc1..b30de1b43d 100644
--- a/haystack/components/generators/chat/openai.py
+++ b/haystack/components/generators/chat/openai.py
@@ -4,6 +4,7 @@
 
 import json
 import os
+from datetime import datetime
 from typing import Any, Callable, Dict, List, Optional, Union
 
 from openai import OpenAI, Stream
@@ -381,6 +382,7 @@ def _convert_streaming_chunks_to_chat_message(self, chunk: Any, chunks: List[Str
             "model": chunk.model,
             "index": 0,
             "finish_reason": chunk.choices[0].finish_reason,
+            "completion_start_time": chunks[0].meta.get("received_at"),  # first chunk received
             "usage": {},  # we don't have usage data for streaming responses
         }
 
@@ -444,6 +446,7 @@ def _convert_chat_completion_chunk_to_streaming_chunk(self, chunk: ChatCompletio
                 "index": choice.index,
                 "tool_calls": choice.delta.tool_calls,
                 "finish_reason": choice.finish_reason,
+                "received_at": datetime.now().isoformat(),
             }
         )
         return chunk_message
diff --git a/haystack/components/generators/hugging_face_api.py b/haystack/components/generators/hugging_face_api.py
index a44ad94575..3ea824cf50 100644
--- a/haystack/components/generators/hugging_face_api.py
+++ b/haystack/components/generators/hugging_face_api.py
@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 from dataclasses import asdict
+from datetime import datetime
 from typing import Any, Callable, Dict, Iterable, List, Optional, Union
 
 from haystack import component, default_from_dict, default_to_dict, logging
@@ -217,18 +218,31 @@ def _stream_and_build_response(
         self, hf_output: Iterable["TextGenerationStreamOutput"], streaming_callback: Callable[[StreamingChunk], None]
     ):
         chunks: List[StreamingChunk] = []
+        first_chunk_time = None
+
         for chunk in hf_output:
             token: TextGenerationOutputToken = chunk.token
             if token.special:
                 continue
-            chunk_metadata = {**asdict(token), **(asdict(chunk.details) if chunk.details else {})}
+
+            chunk_metadata = {
+                **asdict(token),
+                **(asdict(chunk.details) if chunk.details else {}),
+                "received_at": datetime.now().isoformat(),
+            }
+
+            if first_chunk_time is None:
+                first_chunk_time = chunk_metadata["received_at"]
+
             stream_chunk = StreamingChunk(token.text, chunk_metadata)
             chunks.append(stream_chunk)
             streaming_callback(stream_chunk)
+
         metadata = {
             "finish_reason": chunks[-1].meta.get("finish_reason", None),
             "model": self._client.model,
             "usage": {"completion_tokens": chunks[-1].meta.get("generated_tokens", 0)},
+            "completion_start_time": first_chunk_time,
         }
         return {"replies": ["".join([chunk.content for chunk in chunks])], "meta": [metadata]}
 
diff --git a/haystack/components/generators/openai.py b/haystack/components/generators/openai.py
index d2f07f9d85..3a87b8c068 100644
--- a/haystack/components/generators/openai.py
+++ b/haystack/components/generators/openai.py
@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import os
+from datetime import datetime
 from typing import Any, Callable, Dict, List, Optional, Union
 
 from openai import OpenAI, Stream
@@ -255,7 +256,7 @@ def _create_message_from_chunks(
                 "model": completion_chunk.model,
                 "index": 0,
                 "finish_reason": finish_reason,
-                # Usage is available when streaming only if the user explicitly requests it
+                "completion_start_time": streamed_chunks[0].meta.get("received_at"),  # first chunk received
                 "usage": dict(completion_chunk.usage or {}),
             }
         )
@@ -296,12 +297,17 @@ def _build_chunk(chunk: Any) -> StreamingChunk:
         :returns:
             The StreamingChunk.
         """
-        # function or tools calls are not going to happen in non-chat generation
-        # as users can not send ChatMessage with function or tools calls
         choice = chunk.choices[0]
         content = choice.delta.content or ""
         chunk_message = StreamingChunk(content)
-        chunk_message.meta.update({"model": chunk.model, "index": choice.index, "finish_reason": choice.finish_reason})
+        chunk_message.meta.update(
+            {
+                "model": chunk.model,
+                "index": choice.index,
+                "finish_reason": choice.finish_reason,
+                "received_at": datetime.now().isoformat(),
+            }
+        )
         return chunk_message
 
     @staticmethod
diff --git a/releasenotes/notes/add-streaming-completion-timestamp-c0ad3b8698a2d575.yaml b/releasenotes/notes/add-streaming-completion-timestamp-c0ad3b8698a2d575.yaml
new file mode 100644
index 0000000000..2718c6fdcd
--- /dev/null
+++ b/releasenotes/notes/add-streaming-completion-timestamp-c0ad3b8698a2d575.yaml
@@ -0,0 +1,4 @@
+---
+enhancements:
+ - |
+   Added completion_start_time metadata to track time-to-first-token (TTFT) in streaming responses from Hugging Face API and OpenAI (Azure).
diff --git a/test/components/generators/chat/test_hugging_face_api.py b/test/components/generators/chat/test_hugging_face_api.py
index e2158ad6e9..1cd56a178e 100644
--- a/test/components/generators/chat/test_hugging_face_api.py
+++ b/test/components/generators/chat/test_hugging_face_api.py
@@ -1,6 +1,7 @@
 # SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
 #
 # SPDX-License-Identifier: Apache-2.0
+from datetime import datetime
 import os
 from unittest.mock import MagicMock, Mock, patch
 
@@ -503,9 +504,13 @@ def test_live_run_serverless_streaming(self):
         assert isinstance(response["replies"], list)
         assert len(response["replies"]) > 0
         assert [isinstance(reply, ChatMessage) for reply in response["replies"]]
-        assert "usage" in response["replies"][0].meta
-        assert "prompt_tokens" in response["replies"][0].meta["usage"]
-        assert "completion_tokens" in response["replies"][0].meta["usage"]
+
+        response_meta = response["replies"][0].meta
+        assert "completion_start_time" in response_meta
+        assert datetime.fromisoformat(response_meta["completion_start_time"]) <= datetime.now()
+        assert "usage" in response_meta
+        assert "prompt_tokens" in response_meta["usage"]
+        assert "completion_tokens" in response_meta["usage"]
 
     @pytest.mark.integration
     @pytest.mark.skipif(
diff --git a/test/components/generators/chat/test_openai.py b/test/components/generators/chat/test_openai.py
index eb50d92739..63a920a8ec 100644
--- a/test/components/generators/chat/test_openai.py
+++ b/test/components/generators/chat/test_openai.py
@@ -546,6 +546,10 @@ def __call__(self, chunk: StreamingChunk) -> None:
         assert callback.counter > 1
         assert "Paris" in callback.responses
 
+        # check that the completion_start_time is set and valid ISO format
+        assert "completion_start_time" in message.meta
+        assert datetime.fromisoformat(message.meta["completion_start_time"]) < datetime.now()
+
     @pytest.mark.skipif(
         not os.environ.get("OPENAI_API_KEY", None),
         reason="Export an env var called OPENAI_API_KEY containing the OpenAI API key to run this test.",
diff --git a/test/components/generators/test_hugging_face_api.py b/test/components/generators/test_hugging_face_api.py
index 0f4be2f9cb..965cf0cf81 100644
--- a/test/components/generators/test_hugging_face_api.py
+++ b/test/components/generators/test_hugging_face_api.py
@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 import os
 from unittest.mock import MagicMock, Mock, patch
+from datetime import datetime
 
 import pytest
 from huggingface_hub import (
@@ -312,3 +313,25 @@ def test_run_serverless(self):
         assert isinstance(response["meta"], list)
         assert len(response["meta"]) > 0
         assert [isinstance(meta, dict) for meta in response["meta"]]
+
+    @pytest.mark.integration
+    @pytest.mark.skipif(
+        not os.environ.get("HF_API_TOKEN", None),
+        reason="Export an env var called HF_API_TOKEN containing the Hugging Face token to run this test.",
+    )
+    def test_live_run_streaming_check_completion_start_time(self):
+        generator = HuggingFaceAPIGenerator(
+            api_type=HFGenerationAPIType.SERVERLESS_INFERENCE_API,
+            api_params={"model": "HuggingFaceH4/zephyr-7b-beta"},
+            streaming_callback=streaming_callback_handler,
+        )
+
+        results = generator.run("What is the capital of France?")
+
+        assert len(results["replies"]) == 1
+        assert "Paris" in results["replies"][0]
+
+        # Verify completion start time in final metadata
+        assert "completion_start_time" in results["meta"][0]
+        completion_start = datetime.fromisoformat(results["meta"][0]["completion_start_time"])
+        assert completion_start <= datetime.now()
diff --git a/test/components/generators/test_openai.py b/test/components/generators/test_openai.py
index e1d865c95f..816412ee97 100644
--- a/test/components/generators/test_openai.py
+++ b/test/components/generators/test_openai.py
@@ -1,6 +1,7 @@
 # SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
 #
 # SPDX-License-Identifier: Apache-2.0
+from datetime import datetime
 import logging
 import os
 from typing import List
@@ -286,6 +287,9 @@ def __call__(self, chunk: StreamingChunk) -> None:
         assert "gpt-4o-mini" in metadata["model"]
         assert metadata["finish_reason"] == "stop"
 
+        assert "completion_start_time" in metadata
+        assert datetime.fromisoformat(metadata["completion_start_time"]) <= datetime.now()
+
         # unfortunately, the usage is not available for streaming calls
         # we keep the key in the metadata for compatibility
         assert "usage" in metadata and len(metadata["usage"]) == 0


Code review comments:
diff_hunk: @@ -272,9 +275,14 @@ def _run_streaming(self, messages: List[Dict[str, str]], generation_kwargs: Dict
 
             finish_reason = choice.finish_reason
 
-            meta = {}
-            if finish_reason:
-                meta["finish_reason"] = finish_reason
+            meta = {
+                "finish_reason": finish_reason if finish_reason else None,
+                "received_at": datetime.now().isoformat(),
+            }
+            chunks_meta.append(meta)
+
+            if first_chunk_time is None:
+                first_chunk_time = meta["received_at"]
path: haystack/components/generators/chat/hugging_face_api.py
text: @user1:
```suggestion
            meta = {}
            if finish_reason:
                meta["finish_reason"] = finish_reason

            if first_chunk_time is None:
                first_chunk_time = datetime.now().isoformat()
```

Do you see any problems with this simpler approach?