chore(llmobs): fix behavior for running multiple evaluators

ddtrace/llmobs/_evaluators/runner.py

@@ -117,14 +117,9 @@ def periodic(self, _wait_sync=False) -> None:
         try:
             if not _wait_sync:
                 for evaluator in self.evaluators:
-                    self.executor.map(
-                        lambda span_event: evaluator.run_and_submit_evaluation(span_event),
-                        [
-                            span_event
-                            for span_event, span in span_events_and_spans
-                            if self.sampler.sample(evaluator.LABEL, span)
-                        ],
-                    )
+                    for span_event, span in span_events_and_spans:
+                        if self.sampler.sample(evaluator.LABEL, span):
+                            self.executor.submit(evaluator.run_and_submit_evaluation, span_event)
             else:
                 for evaluator in self.evaluators:
                     for span_event, span in span_events_and_spans:

tests/llmobs/_utils.py

@@ -531,28 +531,27 @@ def _llm_span_with_expected_ragas_inputs_in_messages(ragas_inputs=None):
 
 
 class DummyEvaluator:
-    LABEL = "dummy"
-
-    def __init__(self, llmobs_service):
+    def __init__(self, llmobs_service, label="dummy"):
         self.llmobs_service = llmobs_service
+        self.LABEL = label
 
     def run_and_submit_evaluation(self, span):
         self.llmobs_service.submit_evaluation(
             span_context=span,
-            label=DummyEvaluator.LABEL,
+            label=self.LABEL,
             value=1.0,
             metric_type="score",
         )
 
 
-def _dummy_evaluator_eval_metric_event(span_id, trace_id):
+def _dummy_evaluator_eval_metric_event(span_id, trace_id, label=None):
     return LLMObsEvaluationMetricEvent(
         join_on={"span": {"span_id": span_id, "trace_id": trace_id}},
         score_value=1.0,
         ml_app="unnamed-ml-app",
         timestamp_ms=mock.ANY,
         metric_type="score",
-        label=DummyEvaluator.LABEL,
+        label=label or "dummy",
         tags=["ddtrace.version:{}".format(ddtrace.__version__), "ml_app:unnamed-ml-app"],
     )
 

tests/llmobs/test_llmobs_evaluator_runner.py

@@ -59,6 +59,29 @@ def test_evaluator_runner_timed_enqueues_eval_metric(llmobs, mock_llmobs_eval_me
     )
 
 
+@pytest.mark.vcr_logs
+def test_evaluator_runner_multiple_evaluators(llmobs, mock_llmobs_eval_metric_writer):
+    evaluator_runner = EvaluatorRunner(interval=0.01, llmobs_service=llmobs)
+    evaluator_runner.evaluators += [
+        DummyEvaluator(llmobs_service=llmobs, label="1"),
+        DummyEvaluator(llmobs_service=llmobs, label="2"),
+        DummyEvaluator(llmobs_service=llmobs, label="3"),
+    ]
+    evaluator_runner.start()
+
+    evaluator_runner.enqueue({"span_id": "123", "trace_id": "1234"}, DUMMY_SPAN)
+
+    time.sleep(0.1)
+
+    calls = [call[0][0] for call in mock_llmobs_eval_metric_writer.enqueue.call_args_list]
+    sorted_calls = sorted(calls, key=lambda x: x["label"])
+    assert sorted_calls == [
+        _dummy_evaluator_eval_metric_event(span_id="123", trace_id="1234", label="1"),
+        _dummy_evaluator_eval_metric_event(span_id="123", trace_id="1234", label="2"),
+        _dummy_evaluator_eval_metric_event(span_id="123", trace_id="1234", label="3"),
+    ]
+
+
 def test_evaluator_runner_on_exit(mock_writer_logs, run_python_code_in_subprocess):
     env = os.environ.copy()
     pypath = [os.path.dirname(os.path.dirname(os.path.dirname(__file__)))]