DSPy (Beta) - Raindrop

Installation

pip install raindrop-dspy dspy

Quick Start

import dspy
from raindrop_dspy import RaindropDSPy

raindrop = RaindropDSPy(api_key="rk_...", user_id="user-123")

lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=lm)

predict = dspy.Predict("question -> answer")
wrapped = raindrop.wrap(predict)

result = wrapped(question="What is the capital of France?")
print(result.answer)

raindrop.shutdown()

What Gets Traced

The DSPy integration automatically captures:

Module calls — input kwargs, output text (from Prediction result), configured model name
Token usage — prompt and completion tokens from get_lm_usage() or the usage attribute
Finish reason — completion finish reason (e.g. stop, length) extracted from the LM history
Errors — exception type and message captured in event properties, then re-raised

Works with dspy.Predict, dspy.ChainOfThought, and custom dspy.Module subclasses.

Configuration

raindrop = RaindropDSPy(
    api_key="rk_...",                  # Optional: your Raindrop API key (omit to disable telemetry)
    user_id="user-123",                # Optional: associate events with a user
    convo_id="convo-456",              # Optional: conversation/thread ID
    tracing_enabled=True,              # Optional: enable OTEL tracing (default: True)
    bypass_otel_for_tools=True,        # Optional: bypass OTEL for tool calls (default: True)
    debug=True,                        # Optional: enable debug logging (default: False)
)

Parameter	Type	Default	Description
`api_key`	`str \| None`	`None`	Raindrop API key. If omitted, telemetry is disabled.
`user_id`	`str \| None`	`None`	Associate all events with a user.
`convo_id`	`str \| None`	`None`	Group events into a conversation.
`tracing_enabled`	`bool`	`True`	Enable OpenTelemetry tracing.
`bypass_otel_for_tools`	`bool`	`True`	Bypass OTEL instrumentation for tool calls.
`debug`	`bool`	`False`	Enable debug logging (sets logger to `DEBUG` level).

Identifying Users

Use identify() to associate a user with traits after initialization:

raindrop.identify("user-123", traits={"name": "Alice", "plan": "pro"})

The traits parameter accepts a dictionary with string, int, bool, or float values.

Tracking Signals

Use track_signal() to attach user feedback or edits to an AI event:

# Basic signal
raindrop.track_signal(
    event_id="evt-abc",
    name="thumbs_up",
)

# Feedback with sentiment
raindrop.track_signal(
    event_id="evt-abc",
    name="user_rating",
    signal_type="feedback",
    sentiment="POSITIVE",
    comment="Great response!",
)

# Edit signal
raindrop.track_signal(
    event_id="evt-abc",
    name="user_edit",
    signal_type="edit",
    after="The corrected response text",
)

Parameter	Type	Default	Description
`event_id`	`str`	required	The event ID to associate the signal with.
`name`	`str`	required	Signal name (e.g. `"thumbs_up"`, `"user_feedback"`).
`signal_type`	`"default" \| "feedback" \| "edit"`	`"default"`	Signal type.
`timestamp`	`str \| None`	`None`	Optional ISO-8601 timestamp.
`properties`	`dict \| None`	`None`	Optional extra properties.
`attachment_id`	`str \| None`	`None`	Optional attachment identifier.
`comment`	`str \| None`	`None`	Optional comment text.
`after`	`str \| None`	`None`	Optional “after” value for edit signals.
`sentiment`	`"POSITIVE" \| "NEGATIVE" \| None`	`None`	Optional sentiment.

Flushing and Shutdown

Always call shutdown() before your process exits to ensure all telemetry is shipped:

raindrop.flush()     # flush pending data
raindrop.shutdown()  # flush + release resources

Finish Reason Tracking

The integration extracts finish_reason from the DSPy LM history automatically. After each call, it inspects dspy.settings.lm.history for the last response and reads response.choices[0].finish_reason. The value (e.g. stop, length) is included in event properties as dspy.finish_reason.

Token Tracking

Token usage is extracted from the DSPy Prediction result:

get_lm_usage() — returns {model: {prompt_tokens, completion_tokens}}. Tokens are summed across models if multiple LMs are used.
usage attribute — fallback for older DSPy versions. Supports both dict and object formats.

Tokens appear in event properties as ai.usage.prompt_tokens and ai.usage.completion_tokens.

Factory Function

The create_raindrop_dspy factory function is available for backwards compatibility:

from raindrop_dspy import create_raindrop_dspy

raindrop = create_raindrop_dspy(
    api_key="rk_...",
    user_id="user-123",
    tracing_enabled=True,
    bypass_otel_for_tools=True,
)

predict = dspy.Predict("question -> answer")
wrapped = raindrop.wrap(predict)

Async Support

DSPy modules with async forward methods are automatically detected and wrapped:

class AsyncPredict(dspy.Module):
    async def forward(self, question: str) -> dspy.Prediction:
        ...

wrapped = raindrop.wrap(AsyncPredict())
result = await wrapped(question="Hello")

Captured Properties

Each event includes the following properties when available:

Property	Description
`ai.usage.prompt_tokens`	Input token count
`ai.usage.completion_tokens`	Output token count
`dspy.finish_reason`	Completion finish reason (e.g. `stop`, `length`)
`error.type`	Exception class name (on error)
`error.message`	Exception message (on error)

​Installation

​Quick Start

​What Gets Traced

​Configuration

​Identifying Users

​Tracking Signals

​Flushing and Shutdown

​Finish Reason Tracking

​Token Tracking

​Factory Function

​Async Support

​Captured Properties