raindrop-dspy 0.0.7

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

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.

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, project_id=None, tracing_enabled=True, bypass_otel_for_tools=True, disable_auto_instrument=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
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
disable_auto_instrument	bool	True	Library auto-instrumentation is opt-in (see below)
debug	bool	False	Enable debug logging (sets logger to DEBUG level)

Library auto-instrumentation is opt-in

As of 0.0.4, disable_auto_instrument defaults to True: the integration no longer lets Traceloop monkey-patch every LLM client library it recognizes in your process (including the OpenAI client DSPy drives through LiteLLM). The wrapper captures input/output, token usage, and model name directly from the wrapped module's forward() calls and DSPy LM history, so no library patching is needed for full dashboards.

If you specifically want LLM-call-level spans from library instrumentation and have verified compatibility in your environment, opt back in with disable_auto_instrument=False.

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