Skip to main content
DSPy is a framework for algorithmically optimizing LM prompts and weights, especially when LMs are used one or more times within a pipeline. Langtrace has first class support for DSPy, allowing you to capture traces from your DSPy pipelines or agents automatically and analyze them in Langtrace. You can also track experiments and the corresponding metrics and evaluations if you are running DSPy experiments.

​
Setup

  1. Follow the DSPy installation guide to install DSPy.
  2. Install Langtrace’s SDK and initialize the SDK in your code.
  3. Create a project on Langtrace with type DSPy.
  4. Run your DSPy pipeline or agent and view the traces in Langtrace.
  5. To run experiments, follow the conventions below and head over to the Experiments tab in Langtrace to view your experiments.
Important: Follow the steps below for running experiments
For experiments to show up, pass the following additional attributes using the inject_additional_attributes. This way Langtrace knows that you are running an experiment:
  1. (Required) experiment - Experiment name. Ex: experiment 1.
  2. (Optional) description - Some useful description about the experiment.
  3. (Optional) run_id - When you want to associate traces to a specific runs, pass a unique run ID. This is useful when you are running Evaluate() as part of your experiment where the traces specific to the Evaluate() will appear as an individual entry.
from langtrace_python_sdk import inject_additional_attributes

optimized_program = inject_additional_attributes(lambda: optimizer.compile(
    RAG(),
    trainset=trainset,
    max_bootstrapped_demos=3,
    max_labeled_demos=4,
    num_trials=15,
    minibatch_size=2,
    minibatch_full_eval_steps=10,
    minibatch=True,
    requires_permission_to_run=False,
), {'experiment': 'experiment 1', 'description': 'some useful description', 'run_id': 'run_1'})

predictor = inject_additional_attributes(lambda: optimized_program(my_question), {'experiment': 'experiment 1', 'description': 'some useful description', 'run_id': 'run_1'})
Note: The Eval Chart will appear when you run dspy’s Evaluate(). Note: Currently the score ranges it supports are between 0 and 100. So if you have scores that do not fall within this range, it could cause some UI issues.

​
Checkpoints

By default, checkpoints are traced for DSPy pipelines. If you would like to disable it, set the following env var, TRACE_DSPY_CHECKPOINT=false
Checkpoint tracing will increase the latency of executions as the state is serialized. Please disable it in production.

​
Project Type

When creating a project in Langtrace, select the project type as DSPy.
traces

​
Inference Metrics

traces

​
Evaluation Scores

traces

​
Troubleshooting

​
Missing LLM Calls in Traces

If you’re not seeing LLM calls in your Langtrace traces when using DSPy, consider the following:
DSPy implements caching of LLM calls by default. This means that repeated identical calls to the language model will not trigger new API requests, and consequently, won’t generate new traces in Langtrace.To ensure you’re seeing all LLM calls:
  1. Disable caching: If you need to trace every call for debugging purposes, you can disable DSPy’s caching mechanism. Refer to the DSPy documentation for instructions on how to disable caching.
  2. Vary your inputs: If you’re testing, make sure to use different inputs for each run to avoid hitting the cache.
  3. Clear the cache: If you need to re-run the same inputs, consider clearing DSPy’s cache between runs.
  4. Check your DSPy configuration: Ensure that your DSPy setup is correctly configured to use the LLM provider you expect.
If you continue to experience issues after considering these points, please contact our support team for further assistance.Remember to always use caching judiciously in production environments to balance between comprehensive tracing and optimal performance.

​
Grouping of Spans in Trace when using ThreadpoolExecutor in DSPy

If you’re using ThreadpoolExecutor in DSPy to parallelize your modules, you may notice that the spans in the trace are not grouped together. This is because the spans are created in the same thread and are not propagated to the parent span. To resolve this issue,
Please use the provided code snippet as an example to ensure that the current tracing context is properly propagated when executing tasks within a ThreadPoolExecutor. The contextvars module is used to copy the current context and run the optimized CoT module within the copied context. By passing contextvars.copy_context().run to the executor.submit method, the current tracing context is propagated to the child span ensuring that the spans are grouped together in the trace.
import dspy
import contextvars
from dspy.datasets.gsm8k import GSM8K, gsm8k_metric
from dspy.teleprompt import BootstrapFewShot
from concurrent.futures import ThreadPoolExecutor

# flake8: noqa
from langtrace_python_sdk import langtrace, with_langtrace_root_span

langtrace.init(api_key="<your-langtrace-api-key>")

turbo = dspy.OpenAI(model="gpt-3.5-turbo", max_tokens=250)
dspy.settings.configure(lm=turbo)

# Load math questions from the GSM8K dataset
gsm8k = GSM8K()
gsm8k_trainset, gsm8k_devset = gsm8k.train[:10], gsm8k.dev[:10]

class CoT(dspy.Module):
  def __init__(self):
      super().__init__()
      self.prog = dspy.ChainOfThought("question -> answer")

  def forward(self, question):
      return self.prog(question=question)

@with_langtrace_root_span(name="parallel_example")
def example():
  # Set up the optimizer: we want to "bootstrap" (i.e., self-generate) 4-shot examples of our CoT program.
  config = dict(max_bootstrapped_demos=4, max_labeled_demos=4)

  # Optimize! Use the `gsm8k_metric` here. In general, the metric is going to tell the optimizer how well it's doing.
  teleprompter = BootstrapFewShot(metric=gsm8k_metric, **config)
  optimized_cot = teleprompter.compile(CoT(), trainset=gsm8k_trainset)

  questions = [
      "What is the sine of 0?",
      "What is the tangent of 100?",
  ]

  with ThreadPoolExecutor(max_workers=2) as executor:
      futures = [executor.submit(contextvars.copy_context().run, optimized_cot, question=q) for q in questions]

      for future in futures:
          ans = future.result()
          print(ans)


if __name__ == "__main__":
  example()