DSPy

Wrap a DSPy module with neatlogs.wrap() and its module runs (CHAIN), LM calls (LLM), and retrieval (RETRIEVER) are captured. Wrapping a dspy.Module self-roots — the module run becomes the trace root — so a single run renders in the dashboard with no extra code.

Prerequisites

pip install -U "neatlogs[dspy]" dspy

Usage

import os
import neatlogs

neatlogs.init(api_key=os.environ["NEATLOGS_API_KEY"], workflow_name="dspy-demo")

import dspy

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

class QA(dspy.Module):
    def __init__(self):
        super().__init__()
        self.answer = dspy.Predict("question -> answer")

    def forward(self, question: str):
        return self.answer(question=question)

program = neatlogs.wrap(QA())

result = program(question="In one sentence, what is DSPy?")
print(result.answer)

neatlogs.flush()
neatlogs.shutdown()

wrap() works on any DSPy version. If you call a bare predictor (e.g. dspy.Predict(...)) outside a dspy.Module, the LM span self-roots on its own; wrap a dspy.Module (as above) to group a multi-step program under one root.

A runnable end-to-end example lives at examples/sdk_examples/dspy_basic.py in the SDK repo.

Sessions & end-user

Say you invoke your DSPy program a few times over a single back-and-forth and want those runs to hang together as one conversation, tied to the person driving it. Wrap the calls in identify() and the root that your DSPy module opens quietly picks up the session and end-user you named:

# Same session_id every turn → one session; end_user_id attributes it to your user.
with neatlogs.identify(session_id=f"conv_{conversation_id}", end_user_id=user_id):
    answer = program(question=message)

Not every run is a conversation. A standalone workflow — a one-off job, not a back-and-forth — is a single trace with no turns to group. You still attribute it to the customer it ran for, sourcing the id and any metadata from your own user or request object:

def answer_question(question, user):
    # Standalone workflow — attribute to the end-user; do NOT pass session_id.
    with neatlogs.identify(
        end_user_id=str(user.id),
        end_user_metadata={"plan": user.plan},
    ):
        return program(question=question).answer

For the whole picture, see Sessions and End-User Identity.