raindrop-dspy 0.0.3

Raindrop observability integration for DSPy

raindrop-dspy

Raindrop observability integration for DSPy. Wraps DSPy modules (Predict, ChainOfThought, etc.) to automatically capture input fields, output fields, model name, and token usage.

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()

Debug Mode

Enable verbose logging to troubleshoot issues:

raindrop = RaindropDSPy(api_key="rk_...", debug=True)

Factory Function (Legacy)

The create_raindrop_dspy factory function is still 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,
)
wrapped = raindrop.wrap(predict)

What Gets Captured

Data	Description
Input fields	All keyword arguments passed to forward()
Output fields	Extracted from the DSPy Prediction result
Model name	From the active dspy.settings.lm configuration
Token usage	Prompt and completion tokens (when available)
Errors	Exception type and message captured in event properties
Finish reason	Completion finish reason (e.g. stop, length) from the LM history

API Reference

RaindropDSPy(api_key=None, user_id=None, convo_id=None, tracing_enabled=True, bypass_otel_for_tools=True, debug=False)

Create a new Raindrop DSPy integration instance.

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)

Methods

Method	Description
wrap(module)	Instrument a DSPy module (sync and async forward supported)
flush()	Flush buffered events to Raindrop
shutdown()	Flush and shut down the client
identify(user_id, traits=None)	Identify a user with optional traits
track_signal(event_id, name, signal_type="default", ...)	Track a signal on an AI event (e.g. thumbs up/down, feedback)

Identifying Users

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

Tracking Signals

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

Flushing and Shutdown

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

Async Support

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

import dspy
from raindrop_dspy import RaindropDSPy

raindrop = RaindropDSPy(api_key="rk_...")

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

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

Double-Wrap Protection

Calling wrap() on an already-wrapped module is a no-op — the module is returned unchanged.

Development

cd packages/dspy-python
pip install -e ".[dev]"
python -m pytest tests/ -v

License

MIT

Full documentation