Use md5 hash of system instructions as the filename for system instructions.

Author: DylanRussell

util/opentelemetry-util-genai/src/opentelemetry/util/genai/_upload/completion_hook.py

@@ -15,6 +15,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import logging
 import posixpath
 import threading
@@ -152,10 +153,21 @@ def done(future: Future[None]) -> None:
                 )
                 self._semaphore.release()
 
-    def _calculate_ref_path(self) -> CompletionRefs:
+    def _calculate_ref_path(
+        self, system_instruction: list[types.MessagePart]
+    ) -> CompletionRefs:
         # TODO: experimental with using the trace_id and span_id, or fetching
         # gen_ai.response.id from the active span.
-
+        system_instruction_hash = None
+        # Use an md5 hash of the system instructions as a filename, when system instructions are text.
+        if all(isinstance(x, types.Text) for x in system_instruction):
+            md5_hash = hashlib.md5()
+            md5_hash.update(
+                "\n".join(x.content for x in system_instruction).encode(
+                    "utf-8"
+                )
+            )
+            system_instruction_hash = md5_hash.hexdigest()
         uuid_str = str(uuid4())
         return CompletionRefs(
             inputs_ref=posixpath.join(
@@ -166,13 +178,17 @@ def _calculate_ref_path(self) -> CompletionRefs:
             ),
             system_instruction_ref=posixpath.join(
                 self._base_path,
-                f"{uuid_str}_system_instruction.{self._format}",
+                f"{system_instruction_hash or uuid_str}_system_instruction.{self._format}",
             ),
         )
 
     def _do_upload(
         self, path: str, json_encodeable: Callable[[], JsonEncodeable]
     ) -> None:
+        # FileSystem class has this method. Only check for system instructions as that's the only where the filename is a hash.
+        # https://filesystem-spec.readthedocs.io/en/latest/api.html#fsspec.spec.AbstractFileSystem.exists
+        if "_system_instruction" in path and self._fs.exists(path):
+            return
         if self._format == "json":
             # output as a single line with the json messages array
             message_lines = [json_encodeable()]
@@ -213,7 +229,7 @@ def on_completion(
             system_instruction=system_instruction or None,
         )
         # generate the paths to upload to
-        ref_names = self._calculate_ref_path()
+        ref_names = self._calculate_ref_path(system_instruction)
 
         def to_dict(
             dataclass_list: list[types.InputMessage]

util/opentelemetry-util-genai/tests/test_upload.py

@@ -14,10 +14,12 @@
 
 
 # pylint: disable=import-outside-toplevel,no-name-in-module
+import hashlib
 import importlib
 import logging
 import sys
 import threading
+import time
 from contextlib import contextmanager
 from typing import Any
 from unittest import TestCase
@@ -120,13 +122,15 @@ def setUp(self):
         mock_fsspec = self._fsspec_patcher.start()
         self.mock_fs = ThreadSafeMagicMock()
         mock_fsspec.url_to_fs.return_value = self.mock_fs, ""
+        self.mock_fs.exists.return_value = False
 
         self.hook = UploadCompletionHook(
             base_path=BASE_PATH,
             max_size=MAXSIZE,
         )
 
     def tearDown(self) -> None:
+        self.mock_fs.reset_mock()
         self.hook.shutdown()
         self._fsspec_patcher.stop()
 
@@ -162,6 +166,46 @@ def test_upload_then_shutdown(self):
             "should have uploaded 3 files",
         )
 
+    def test_system_insruction_is_hashed_to_avoid_reupload(self):
+        system_instructions = [
+            types.Text(content="You are a helpful assistant."),
+            types.Text(content="You will do your best."),
+        ]
+        md5_hash = hashlib.md5()
+        md5_hash.update(
+            "\n".join(x.content for x in system_instructions).encode("utf-8")
+        )
+        expected_hash = md5_hash.hexdigest()
+        record = LogRecord()
+        self.hook.on_completion(
+            inputs=[],
+            outputs=[],
+            system_instruction=system_instructions,
+            log_record=record,
+        )
+        # Wait a bit for file upload to finish..
+        time.sleep(0.5)
+        self.mock_fs.exists.return_value = True
+        self.hook.on_completion(
+            inputs=[],
+            outputs=[],
+            system_instruction=system_instructions,
+            log_record=record,
+        )
+        # all items should be consumed
+        self.hook.shutdown()
+
+        self.assertEqual(
+            self.mock_fs.open.call_count,
+            1,
+            "should have uploaded 1 file",
+        )
+        assert record.attributes is not None
+        self.assertEqual(
+            record.attributes["gen_ai.system_instructions_ref"].split("/")[-1],
+            f"{expected_hash}_system_instruction.json",
+        )
+
     def test_upload_when_inputs_outputs_empty(self):
         record = LogRecord()
         self.hook.on_completion(