Trace DSPy programs and ReAct agents with Maple: OpenInference DSPy instrumentor with GenAI dual-write, a DSPy callback for tokens, cost, tool names and agent spans, using_session for one session per conversation, and thread context for dspy.Parallel. Triggers on 'trace my dspy agent', 'add Maple to dspy', 'agent sessions for dspy', 'OpenTelemetry for dspy'.
72
91%
Does it follow best practices?
Run evals on this skill
Adds up to 20 points to the overall score
Passed
No findings from the security scan
Goal: every conversation with the user's DSPy program shows up in Maple Agent Sessions as one session, with one turn per call to the program, a transcript, model calls with tokens and cost, tool calls with arguments, results and failures, and one lane per worker module.
Spans come from openinference-instrumentation-dspy. It records no tokens, no tool names, no agent spans and no session id; the steps below add all four. Do not skip any step.
pyproject.toml / requirements*.txt / uv.lock. Require dspy>=3.4. On 3.3 or older, tell the user this skill targets 3.4 and ask before upgrading.dspy.LM("provider/model", ...). Note any engine=, cache=, callbacks=, disable_history, max_history_size.TracerProvider(, set_tracer_provider, opentelemetry-instrument, logfire.configure, mlflow.dspy.autolog, phoenix.otel.register, LiteLLMInstrumentor, OpenAIInstrumentor.
TracerProvider: reuse it. Add the OTLP exporter to it and pass it to instrument(). Never create a second provider.LiteLLMInstrumentor / OpenAIInstrumentor from OpenInference: remove them (they double-count model calls on the LiteLLM engine and record nothing on the native one).mlflow.dspy.autolog(): leave it if the user wants MLflow too, but it is not the Maple path.dspy.Parallel, dspy.Evaluate, ThreadPoolExecutor, threading.Thread around module calls.dspy.streamify(...) calls and where they run relative to dspy.configure (see Streaming in Step 3).dspy.Module subclass with an attribute named history (see "Do not").https://ingest.maple.dev. EU: https://ingest.eu.maple.dev.Authorization=Bearer <key>.MAPLE_TEST (ingest accepts and discards it) and tell the user to replace it with a key from Settings → Ingestion.maple_sk_ key in browser code..env, settings module, secret manager). If there is none, inline is acceptable: ingest keys are write-only..env (load_dotenv()): call it at the top of tracing.py, before the provider is built. Otherwise the exporter silently targets localhost:4318 with no key.tracing.py, after any load_dotenv(), when OTEL_EXPORTER_OTLP_HEADERS is unset, log one warning (logging.getLogger(__name__).warning("OTEL_EXPORTER_OTLP_HEADERS (Maple ingest key) is not set; Maple telemetry export is disabled")) and skip the provider and exporter setup. Never raise or exit over the key, and never send a header without one (opaque 401).pip install "dspy>=3.4" "openinference-instrumentation-dspy>=0.1.45" "openinference-instrumentation>=0.1.66" \
"opentelemetry-sdk>=1.45" "opentelemetry-exporter-otlp-proto-http>=1.45" \
"opentelemetry-instrumentation-threading>=0.66b0"Use the repo's package manager (uv add, poetry add, requirements file).
Environment (base URL, no /v1/traces; the exporter appends it):
OTEL_SERVICE_NAME=<service name>
OTEL_RESOURCE_ATTRIBUTES=deployment.environment.name=<env>
OTEL_EXPORTER_OTLP_ENDPOINT=https://ingest.maple.dev
OTEL_EXPORTER_OTLP_HEADERS="Authorization=Bearer <key>"
OTEL_EXPORTER_OTLP_PROTOCOL=http/protobufCreate tracing.py:
from openinference.instrumentation import TraceConfig
from openinference.instrumentation.dspy import DSPyInstrumentor
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.threading import ThreadingInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
provider = TracerProvider() # reads OTEL_SERVICE_NAME and OTEL_RESOURCE_ATTRIBUTES
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(provider)
DSPyInstrumentor().instrument(tracer_provider=provider, config=TraceConfig(enable_genai_semconv=True))
ThreadingInstrumentor().instrument()enable_genai_semconv=True is required.ThreadingInstrumentor is required whenever modules run in threads (dspy.Parallel, Evaluate, executors). Without it every worker is an orphan trace with no session.tracing as the first import of every entry point (web app, CLI, worker), before the DSPy program modules.Create maple_dspy.py exactly as below (it fills the instrumentor's gaps from DSPy's callback hooks, which run inside the instrumentor's spans):
import json
import dspy
from dspy.utils.callback import BaseCallback
from openinference.instrumentation import TraceConfig
from opentelemetry import trace
# Same switches as the instrumentor: OPENINFERENCE_HIDE_INPUTS / OPENINFERENCE_HIDE_OUTPUTS.
_config = TraceConfig()
def _message(role, values):
text = "\n".join(v for v in values if isinstance(v, str))
return json.dumps([{"role": role, "parts": [{"type": "text", "content": text}]}]) if text else None
class MapleCallback(BaseCallback):
def __init__(self):
self._agents = set()
self._lms = {}
def on_module_start(self, call_id, instance, inputs):
if type(instance).__module__.startswith("dspy."):
return # Predict, ChainOfThought, ReAct: building blocks, not agents
self._agents.add(call_id)
span = trace.get_current_span()
span.set_attribute("gen_ai.operation.name", "invoke_agent")
span.set_attribute("gen_ai.agent.name", type(instance).__name__)
user = _message("user", [*inputs.get("args", ()), *inputs.get("kwargs", {}).values()])
if user and not _config.hide_inputs:
span.set_attribute("gen_ai.input.messages", user)
def on_module_end(self, call_id, outputs, exception):
if call_id not in self._agents:
return
self._agents.discard(call_id)
if isinstance(outputs, dspy.Prediction) and not _config.hide_outputs:
reply = _message("assistant", [v for k, v in outputs.items() if k != "reasoning"])
if reply:
trace.get_current_span().set_attribute("gen_ai.output.messages", reply)
def on_tool_start(self, call_id, instance, inputs):
span = trace.get_current_span()
span.set_attribute("gen_ai.tool.name", instance.name)
span.set_attribute("gen_ai.tool.description", instance.desc or "")
if not _config.hide_inputs:
span.set_attribute("gen_ai.tool.call.arguments", json.dumps(inputs.get("kwargs", {}), default=str))
def on_lm_start(self, call_id, instance, inputs):
self._lms[call_id] = instance
def on_lm_end(self, call_id, outputs, exception):
lm = self._lms.pop(call_id, None)
# The LM's history holds the provider response. Threads share the LM, so match ours by identity.
entry = next((e for e in reversed(lm.history[-16:]) if e["outputs"] is outputs), None) if lm else None
if entry is None or getattr(entry["response"], "cache_hit", False):
return # history is off, or a cache hit that cost nothing
usage = entry["usage"] or {}
attributes = {
"gen_ai.response.id": getattr(entry["response"], "id", None),
"gen_ai.response.model": entry.get("response_model"),
"gen_ai.usage.input_tokens": usage.get("prompt_tokens"),
"gen_ai.usage.output_tokens": usage.get("completion_tokens"),
"gen_ai.usage.cache_read.input_tokens": (usage.get("prompt_tokens_details") or {}).get("cached_tokens"),
"gen_ai.usage.reasoning.output_tokens": (usage.get("completion_tokens_details") or {}).get("reasoning_tokens"),
"gen_ai.usage.cost": entry.get("cost"),
}
trace.get_current_span().set_attributes({k: v for k, v in attributes.items() if v is not None})Register it where the app configures DSPy, keeping existing callbacks:
import tracing # first
import dspy
from maple_dspy import MapleCallback
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"), callbacks=[MapleCallback()]) # append to any existing callbacks listdspy.configure(callbacks=...) or dspy.context(callbacks=...) must include the MapleCallback instance too, or it is replaced.dspy.ReAct / dspy.ChainOfThought called directly, wrap it in a small dspy.Module subclass named after the agent. Only user-defined module classes become agents (invoke_agent, gen_ai.agent.name = class name).Maple reads session.id for DSPy. Wrap every call to the program in using_session with the app's own stored conversation id:
from openinference.instrumentation import using_session
def handle_message(conversation_id: str, question: str, turns: list[dict]) -> str:
with using_session(conversation_id):
answer = assistant(question=question, conversation=dspy.History(messages=turns)).answer
turns.append({"question": question, "answer": answer})
return answerusing_session is context-based).Streaming (dspy.streamify):
stream_assistant = dspy.streamify(assistant, stream_listeners=[dspy.streaming.StreamListener("answer")])
async def stream_message(conversation_id: str, question: str, turns: list[dict]):
with using_session(conversation_id):
async for chunk in stream_assistant(question=question, conversation=dspy.History(messages=turns)):
if isinstance(chunk, dspy.streaming.StreamResponse):
yield chunk.chunk
elif isinstance(chunk, dspy.Prediction):
turns.append({"question": question, "answer": chunk.answer})dspy.streamify(...) only after dspy.configure(callbacks=[MapleCallback()]): it snapshots the callback list when called. A streamified program created earlier (e.g. at import time) streams without the callback: no tokens, no tool names.using_session must wrap the loop that consumes the stream (inside the generator handed to the SSE/StreamingResponse), not only the stream_assistant(...) call: the program starts on the first iteration.gen_ai.response.id (DSPy's native engine drops it) and no TTFT attribute. Expected.OPENINFERENCE_HIDE_INPUTS=true and OPENINFERENCE_HIDE_OUTPUTS=true, set before tracing.py / maple_dspy.py are imported. The callback honours both.dspy.ReAct(tools=[...]) or dspy.Tool(...) get <name>.__call__ spans. A raised exception marks the tool span ERROR with the message; Maple counts it. Do not catch exceptions inside the tool just to return an error string: that hides the failure.gen_ai.tool.description.WeatherWorker, BudgetWorker). Two instances of one class share one lane.dspy.Parallel is fine once ThreadingInstrumentor runs. multiprocessing is not covered; each process starts its own trace.provider.shutdown() in a finally at the end of main.provider.force_flush() before returning / after each run.from tracing import provider
try:
run()
finally:
provider.shutdown()Run one conversation of 2-3 messages with the same conversation id (one using a tool), with cache=False on the LM so every call hits the provider. No scriptable entry point (server, UI, REPL only): write a small driver for this run (one conversation id, 2+ turns, one tool call, flush before exit).
Without Maple access (MAPLE_TEST, no MCP): the run must exit with no export errors on stderr (Failed to export, 401 lines) AND a local exporter (provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter())) in a scratch run) must show <YourModule>.forward, LM.__call__ and <tool>.__call__ spans carrying session.id. Silence alone proves nothing (no spans is silent too). With the Maple MCP: list_agent_sessions with search=<conversation id> returns one row.
Then check in Maple Agent Sessions (https://app.maple.dev/agent-sessions, or app.eu.maple.dev), or with the Maple MCP (list_agent_sessions, get_agent_session):
<YourModule>.forward, titled with the user's question.LM.__call__, each with model, input and output tokens, and cost where DSPy knows the price. The LLM call count equals the real number of model calls (not double).LM.__call__ spans have tokens.[[ ## field ## ]] prompt format is expected).<tool>.__call__ with gen_ai.tool.name, arguments and results. finish.__call__ (ReAct's end-of-loop tool) is expected and counts as a tool call.dspy.Parallel, all workers are in the same trace as the orchestrator, none orphaned.401 ingest_unauthorized / "Invalid ingest key" with a key you trust: keys are region-bound, so it likely belongs to the other region; try the other endpoint.
OPENINFERENCE_ENABLE_GENAI_SEMCONV=true set before instrument() runs is equivalent to TraceConfig(enable_genai_semconv=True). The GenAI dual-write never overwrites a key already set, so the callback's values win.LM.__call__ sits under Predict.forward, Predict(StringSignature).forward and ChatAdapter.__call__; those are DSPy steps, not extra model calls.dspy.ReAct catches tool exceptions and hands Execution error in <tool>: ... back to the model. The tool span is ERROR and counted as failed; the ReAct.forward span and the user's module span stay OK, and the program's return value doesn't reveal the failure.gen_ai.tool.call.id: ReAct asks the model for the next tool as text fields (next_tool_name, next_tool_args), not through the provider's tool-calling API.ChatAdapter can't parse a reply, DSPy retries with JSONAdapter: one Predict span holds two adapter spans, each with its own LM.__call__. Both calls happened and both are billed.cost: on the lm15 engine from DSPy's bundled model metadata, on the LiteLLM engine LiteLLM's response_cost. Maple never prices tokens; a model DSPy can't price shows as unpriced.LM with a custom engine= reports whatever usage that engine puts on its response.dspy.LM(..., engine="litellm")); that is where an extra LiteLLM/OpenAI instrumentor would double-count.dspy.Parallel copies only DSPy's settings into its worker threads, not the OpenTelemetry context; ThreadingInstrumentor is what carries it.OPENINFERENCE_HIDE_INPUT_TEXT, OPENINFERENCE_HIDE_OUTPUT_TEXT, OPENINFERENCE_HIDE_LLM_INVOCATION_PARAMETERS) redact parts of each model message; the callback's agent messages follow only OPENINFERENCE_HIDE_INPUTS / OPENINFERENCE_HIDE_OUTPUTS.dspy.Module attribute history: DSPy appends LM calls to it and a dspy.History there crashes every model call (TypeError: object of type 'History' has no len()). Pass dspy.History as an input field.disable_history=True or max_history_size=0: the callback reads token usage from LM history.opentelemetry-instrumentation-genai-dspy as the Maple path: 1.2b0 records no model calls and is not identified as DSPy.MIPROv2, GEPA, BootstrapFewShot) or dspy.Evaluate under the production service name; they make hundreds of calls.MapleTechLabs/maple
27a645c
If you maintain this skill, you can claim it as your own. Once claimed, you can manage eval scenarios, bundle related skills, attach documentation or rules, and ensure cross-agent compatibility.