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
    project_id="support-prod",         # Optional: route events to a specific project (slug)
    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.
`project_id`	`str \| None`	`None`	Route events to a specific project (slug); omit for the default Production project.
`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).

Projects

Route events to a specific project by passing its slug as project_id:

raindrop = RaindropDSPy(
    api_key="rk_...",
    project_id="support-prod",
)

project_id sets the X-Raindrop-Project-Id header on every event. Omit it (or pass "default") to use your org’s default Production project, which is the existing behavior. The same option is accepted by the create_raindrop_dspy(...) factory. Invalid slugs are ignored with a warning and no header is sent.

Multiple projects in one process

Available in raindrop-ai>=0.0.56. When one service runs several DSPy modules that should report to different projects, create one RaindropDSPy wrapper per project. Each wrapper owns its own raindrop.Raindrop client, so the two route independently — there is no shared module-level state:

from raindrop_dspy import RaindropDSPy

# One long-lived wrapper per project, created at startup and reused.
rd_support = RaindropDSPy(api_key="rk_...", project_id="support-prod")
rd_billing = RaindropDSPy(api_key="rk_...", project_id="billing-prod")

support_module = rd_support.wrap(support_module)   # -> support-prod
billing_module = rd_billing.wrap(billing_module)   # -> billing-prod

rd_support.flush()
rd_billing.flush()

Each wrapper owns its configuration and delivery pipeline, so modules handled by different wrappers route independently. To share a single client across wrappers (or with the module-level API), construct a raindrop.Raindrop yourself and pass it via client=:

from raindrop import Raindrop

client = Raindrop(api_key="rk_...", project_id="support-prod")
raindrop = RaindropDSPy(client=client)

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

​Projects

​Multiple projects in one process

​Identifying Users

​Tracking Signals

​Flushing and Shutdown

​Finish Reason Tracking

​Token Tracking

​Factory Function

​Async Support

​Captured Properties