Fix broken Otel propagation in dspy ParallelExecutor

[TomeHirata]

Currently, dspy.Parallel has an issue with the tree structure of MLflow tracing gets broken when dspy.Parallel is used. This is because Dspy doesn't copy the parent context when workers are executed in parallel. It is intentional not to copy the entire contextvars in order to maintain different contextvar across different threads. This PR manually propagates the Otel Context to child threads when ParallelExecutor is executed to fix the tracing issue.

class ParallelExample(dspy.Module):
  def __init__(self):
      self.keyword_extractor = dspy.Predict("text -> keywords")
      self.sentiment_extractor = dspy.Predict("text -> sentiment")
      self.parallel = dspy.Parallel(num_threads=2)

  def forward(self, text):
      input_example = dspy.Example(text=text).with_inputs("text")

      results = self.parallel([
          (self.keyword_extractor, input_example),
          (self.sentiment_extractor, input_example)
      ])

      return results

# Usage
processor = ParallelExample()
result = processor(text="I love using DSPy! It makes AI programming so much easier.")

print(result)

[Copilot]

Pull Request Overview

This PR ensures OpenTelemetry context is propagated to threads in ParallelExecutor, fixes MLflow trace hierarchy, and adds a test to verify the behavior.

• Introduce _with_otel_context decorator and apply it to the worker function in ParallelExecutor
• Add test_otel_context_propagation to cover OpenTelemetry context forwarding
• Update pyproject.toml to include OpenTelemetry dependencies

Reviewed Changes

Copilot reviewed 3 out of 4 changed files in this pull request and generated 1 comment.

File	Description
tests/utils/test_parallelizer.py	Added OpenTelemetry context propagation test
pyproject.toml	Added opentelemetry-api and opentelemetry-sdk
dspy/utils/parallelizer.py	Implemented _with_otel_context and applied decorator to worker

Comments suppressed due to low confidence (1)

tests/utils/test_parallelizer.py:64

• The test references ParallelExecutor but does not import it, causing a NameError. Add: from dspy.utils.parallelizer import ParallelExecutor.

@pytest.mark.skipif(not importlib.util.find_spec("opentelemetry"), reason="OpenTelemetry not installed")