maple-agent-tracing-dspy

Maple agent tracing: DSPy

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.

Step 0: Detect versions and existing setup

Read 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.
Find how LMs are built: dspy.LM("provider/model", ...). Note any engine=, cache=, callbacks=, disable_history, max_history_size.
Search for an existing OpenTelemetry setup: TracerProvider(, set_tracer_provider, opentelemetry-instrument, logfire.configure, mlflow.dspy.autolog, phoenix.otel.register, LiteLLMInstrumentor, OpenAIInstrumentor.
- An existing 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.
Find where the program is called per user message (HTTP handler, CLI loop, worker). That is where the session id goes.
Find dspy.Parallel, dspy.Evaluate, ThreadPoolExecutor, threading.Thread around module calls.
Find dspy.streamify(...) calls and where they run relative to dspy.configure (see Streaming in Step 3).
Check for a dspy.Module subclass with an attribute named history (see "Do not").

Step 1: Ingest key and region

US: https://ingest.maple.dev. EU: https://ingest.eu.maple.dev.
Header: Authorization=Bearer <key>.
Key in the user's prompt: use it. No key: use the literal MAPLE_TEST (ingest accepts and discards it) and tell the user to replace it with a key from Settings → Ingestion.
Never put a private maple_sk_ key in browser code.
Follow the repo's existing secret/env convention (.env, settings module, secret manager). If there is none, inline is acceptable: ingest keys are write-only.
App loads .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.
Key missing: instrumentation must never crash or block the app. In 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).

Step 2: Install and initialize

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/protobuf

Create 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.
Import 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})

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 list

Every later dspy.configure(callbacks=...) or dspy.context(callbacks=...) must include the MapleCallback instance too, or it is replaced.
If the app's program is a bare 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).

Step 3: One session per conversation

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 answer

Use the id the app already stores the chat under. Never a new UUID per request, never a constant.
For a one-shot job (batch, CLI run, orchestration), mint one id per run and wrap the whole run.
Async handlers work unchanged (using_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})

Call 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.
Streamed calls get tokens and cost; they have no gen_ai.response.id (DSPy's native engine drops it) and no TTFT attribute. Expected.

Step 4: Content

Default: prompts, replies, tool arguments and results are captured. Maple shows them as the transcript.
To turn content off: OPENINFERENCE_HIDE_INPUTS=true and OPENINFERENCE_HIDE_OUTPUTS=true, set before tracing.py / maple_dspy.py are imported. The callback honours both.
If the user mentions PII or compliance, ask whether to disable content; do not decide silently.

Step 5: Tools, errors, sub-agents

Tools need no changes: plain functions passed to 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.
Give tools real docstrings; the callback records them as gen_ai.tool.description.
Sub-agents: DSPy's idiom is an orchestrator module calling worker modules. Give each worker its own class with a descriptive name (WeatherWorker, BudgetWorker). Two instances of one class share one lane.
Parallel workers: dspy.Parallel is fine once ThreadingInstrumentor runs. multiprocessing is not covered; each process starts its own trace.

Step 6: Flush

Long-running servers: nothing to add; the provider flushes on normal exit.
Scripts, CLIs, jobs: provider.shutdown() in a finally at the end of main.
Serverless handlers and notebooks: provider.force_flush() before returning / after each run.

from tracing import provider

try:
    run()
finally:
    provider.shutdown()

Step 7: Verify

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

Exactly one session for the conversation, id = the conversation id, framework DSPy. A second conversation is a second session.
One turn per program call; each turn's root is <YourModule>.forward, titled with the user's question.
Model calls named 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).
If the app streams: the streamed turn is in the same session and its LM.__call__ spans have tokens.
Transcript non-empty (DSPy's [[ ## field ## ]] prompt format is expected).
Tool calls named <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.
A tool that raised is counted as failed (session check Tool availability fails with the exception); no successful tool or model call is marked failed.
Worker modules appear as separate agents/lanes; with dspy.Parallel, all workers are in the same trace as the orchestrator, none orphaned.
The process exited cleanly and the last turn is present (flush ran).

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.

Known behaviours (expected, nothing to fix; explain them if the user asks)

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.
Each 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.
Tool spans have no 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.
When 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 is DSPy's estimate from each history entry's 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.
An LM with a custom engine= reports whatever usage that engine puts on its response.
Anthropic models run on the LiteLLM engine by default (as does 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.
Narrower content switches (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.
With content hidden the session still shows turns, model and tool calls, tokens and failures, with an empty transcript and no tool arguments or results.

Do not

Do not name a 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.
Do not set disable_history=True or max_history_size=0: the callback reads token usage from LM history.
Do not use opentelemetry-instrumentation-genai-dspy as the Maple path: 1.2b0 records no model calls and is not identified as DSPy.
Do not trace optimizer runs (MIPROv2, GEPA, BootstrapFewShot) or dspy.Evaluate under the production service name; they make hundreds of calls.