[OPIK-770] Fix haystack integration (#1070)

* Put error details closer to the beginning of error messages to simplify sentry logs review

* Fix haystack bug

* Remove version restriction in integration test for haystack

* Fix lint errors

sdks/python/src/opik/integrations/haystack/converters.py

@@ -0,0 +1,31 @@
+import functools
+import importlib.metadata
+from typing import Any, Dict
+
+import haystack.dataclasses
+import haystack.version
+
+from opik import semantic_version
+
+
+def convert_message_to_openai_format(
+    message: haystack.dataclasses.ChatMessage,
+) -> Dict[str, Any]:
+    if _haystack_version_less_than_2_9_0():
+        # 2.8.1 and less use _convert_message_to_openai_format function
+        from haystack.components.generators.openai import (
+            _convert_message_to_openai_format,
+        )
+
+        return _convert_message_to_openai_format(message=message)
+    else:
+        # 2.9.0 introduced to_openai_dict_format
+        return message.to_openai_dict_format()
+
+
+@functools.lru_cache
+def _haystack_version_less_than_2_9_0() -> bool:
+    haystack_version = importlib.metadata.version("haystack-ai")
+    result = semantic_version.SemanticVersion.parse(haystack_version) < "2.9.0"  # type: ignore
+
+    return result

sdks/python/src/opik/integrations/haystack/opik_tracer.py

@@ -4,14 +4,14 @@
 from typing import Any, Dict, Iterator, List, Optional, Union
 
 from haystack import logging
-from haystack.components.generators import openai_utils
 from haystack import dataclasses as haystack_dataclasses
 from haystack import tracing
 from haystack.tracing import utils as tracing_utils
 
 import opik
 from opik.api_objects import span as opik_span
 from opik.api_objects import trace as opik_trace
+from . import converters
 
 logger = logging.getLogger(__name__)
 
@@ -90,21 +90,21 @@ def set_content_tag(self, key: str, value: Any) -> None:
         if key.endswith(".input"):
             if "messages" in value:
                 messages = [
-                    openai_utils._convert_message_to_openai_format(m)
-                    for m in value["messages"]
+                    converters.convert_message_to_openai_format(message)
+                    for message in value["messages"]
                 ]
                 self._span.update(input={"input": messages})
             else:
                 self._span.update(input={"input": value})
         elif key.endswith(".output"):
             if "replies" in value:
                 if all(
-                    isinstance(r, haystack_dataclasses.ChatMessage)
-                    for r in value["replies"]
+                    isinstance(reply, haystack_dataclasses.ChatMessage)
+                    for reply in value["replies"]
                 ):
                     replies = [
-                        openai_utils._convert_message_to_openai_format(m)
-                        for m in value["replies"]
+                        converters.convert_message_to_openai_format(message)
+                        for message in value["replies"]
                     ]
                 else:
                     replies = value["replies"]

sdks/python/src/opik/logging_messages.py

@@ -23,7 +23,7 @@
 FAILED_TO_PARSE_OPENAI_STREAM_CONTENT = "Failed to parse openai Stream content. %s"
 
 FAILED_TO_PROCESS_MESSAGE_IN_BACKGROUND_STREAMER = (
-    "Failed to process %s.\nContent: %s,\nError: %s"
+    "Failed to process %s. Error: %s,\nContent: %s"
 )
 
 HALLUCINATION_DETECTION_FAILED = "Failed hallucination detection"

sdks/python/src/opik/message_processing/message_processors.py

@@ -57,15 +57,15 @@ def process(self, message: messages.BaseMessage) -> None:
             LOGGER.error(
                 logging_messages.FAILED_TO_PROCESS_MESSAGE_IN_BACKGROUND_STREAMER,
                 message_type.__name__,
-                message,
                 str(exception),
+                message,
             )
         except Exception as exception:
             LOGGER.error(
                 logging_messages.FAILED_TO_PROCESS_MESSAGE_IN_BACKGROUND_STREAMER,
                 message_type.__name__,
-                message,
                 str(exception),
+                message,
                 exc_info=True,
             )
 

sdks/python/tests/library_integration/haystack/requirements.txt

@@ -1 +1 @@
-haystack-ai<2.9.0  # 2.9.0 broke our imports, we need to fix it! https://github.com/comet-ml/opik/issues/1047
+haystack-ai