SuperAPI AI Directory
Partner with us中文AI newsFavorites
Skillsmaple-agent-tracing-haystack

maple-agent-tracing-haystack

Agents & models

View repository

Updated 10-16 00:05

About this skill

Maple agent tracing: Haystack

Goal: one conversation = one Maple Agent Session, with transcript, model calls, tool calls (failures marked), tokens and (on OpenRouter) cost.

Step 0: Detect

  • haystack-ai version: python -c "import haystack; print(haystack.__version__)". Need >=3.0; guide verified on 3.2. For 2.x, stop and tell the user this skill targets Haystack 3.
  • Find where Agents/Pipelines run (Agent(, Pipeline(, .run(, AgentTool(, PipelineTool() and where a chat/thread id is available per request.
  • Existing OTel: grep for TracerProvider(, set_tracer_provider, configure_otel, logfire.configure, opentelemetry-instrument. If a provider exists, reuse it: only add the Maple exporter (if not already exporting to Maple) and enable_tracing(...). Never create a second provider.
  • Remove/skip HaystackInstrumentor().instrument() (OpenInference) and OpenLLMetry's Haystack instrumentor if present: they double every model call. Remove any existing enable_tracing(OpenTelemetryTracer(...)) or OpenTelemetryConnector component and replace with Step 2.

Step 1: Key and region

  • US endpoint https://ingest.maple.dev; EU endpoint https://ingest.eu.maple.dev. Header Authorization=Bearer <key>. Protocol http/protobuf.
  • Key given in the prompt: use it. No key: use the literal MAPLE_TEST (ingest accepts and discards it) and tell the user to replace it with their key from Settings → Ingestion.
  • Private maple_sk_ keys never go in browser code.
  • Follow the repo's secret/env convention (.env, settings module, deployment env). If none exists, inline is acceptable: ingest keys are write-only.
  • Building the header in code from an env var: when it is unset, log one warning (MAPLE_INGEST_KEY is not set; Maple telemetry export is disabled) and skip the Maple exporter so the app runs normally. Never raise or exit over the key, and never send Bearer None (opaque 401) or hit a bare KeyError on import. Or inline the key when the repo has no env convention.
  • 401 ingest_unauthorized / "Invalid ingest key" with a key you trust: keys are region-bound, so it usually belongs to the other region. Try the other endpoint.

Step 2: Install and init

pip install "haystack-ai>=3.2" "opentelemetry-haystack>=1.0" "opentelemetry-sdk>=1.45" "opentelemetry-exporter-otlp-proto-http>=1.45"

Use the repo's package manager (uv add, poetry add, requirements file). Write this file verbatim as maple_haystack.py in the app's package:

"""Haystack tracer that adds the OpenTelemetry GenAI attributes Maple reads."""

import json
import logging
from collections.abc import Iterator
from contextlib import contextmanager
from contextvars import ContextVar
from typing import Any

from haystack.dataclasses import ChatMessage
from haystack_integrations.tracing.opentelemetry import OpenTelemetrySpan, OpenTelemetryTracer
from opentelemetry import trace
from opentelemetry.trace import StatusCode

logger = logging.getLogger(__name__)
_conversation_id: ContextVar[str | None] = ContextVar("maple_conversation_id", default=None)


@contextmanager
def conversation(conversation_id: str) -> Iterator[None]:
    """Every Haystack run inside this block joins the same Maple session."""
    token = _conversation_id.set(conversation_id)
    try:
        yield
    finally:
        _conversation_id.reset(token)


def _messages(messages: list[ChatMessage]) -> str:
    out = []
    for m in messages:
        parts: list[dict[str, Any]] = [{"type": "text", "content": t} for t in m.texts]
        parts += [{"type": "tool_call", "id": c.id, "name": c.tool_name, "arguments": c.arguments} for c in m.tool_calls]
        parts += [{"type": "tool_call_response", "id": r.origin.id, "response": r.result} for r in m.tool_call_results]
        out.append({"role": m.role.value, "parts": parts})
    return json.dumps(out, default=str)


class MapleSpan(OpenTelemetrySpan):
    def __init__(self, span: trace.Span, operation: str | None, content: bool) -> None:
        super().__init__(span)
        self._operation = operation
        self._content = content

    def set_content_tag(self, key: str, value: Any) -> None:
        try:
            if self._operation == "chat":
                self._chat(key, value)
            elif self._operation == "execute_tool":
                self._tool(key, value)
        except Exception:  # a tracing bug must never fail the agent run
            logger.exception("maple_haystack: could not map %s", key)
        if self._content:
            self.set_tag(key, value)

    def _chat(self, key: str, value: Any) -> None:
        if key.endswith(".input") and self._content:
            messages = value["messages"]
            system = [{"type": "text", "content": m.text} for m in messages if m.is_from("system")]
            if system:
                self._span.set_attribute("gen_ai.system_instructions", json.dumps(system))
            self._span.set_attribute("gen_ai.input.messages", _messages([m for m in messages if not m.is_from("system")]))
        elif key.endswith(".output"):
            replies = value["replies"]
            meta = replies[0].meta
            usage = meta.get("usage") or {}
            prompt_details = usage.get("prompt_tokens_details") or {}
            attributes = {
                "gen_ai.response.model": meta.get("model"),
                "gen_ai.response.finish_reasons": meta.get("finish_reason"),
                "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": prompt_details.get("cached_tokens"),
                "gen_ai.usage.cache_write.input_tokens": prompt_details.get("cache_write_tokens"),
                "gen_ai.usage.reasoning.output_tokens": (usage.get("completion_tokens_details") or {}).get("reasoning_tokens"),
                "gen_ai.usage.cost": usage.get("cost"),  # OpenRouter prices every call; other providers leave this out
            }
            if self._content:
                attributes["gen_ai.output.messages"] = _messages(replies)
            self._span.set_attributes({k: v for k, v in attributes.items() if v is not None})

    def _tool(self, key: str, value: Any) -> None:
        if key.endswith(".output") and isinstance(value, dict) and "error" in value:
            # Haystack records a failed tool as {"error": ...} and leaves the span status unset
            self._span.set_status(StatusCode.ERROR, str(value["error"]) if self._content else "Tool invocation failed")
            self._span.set_attribute("error.type", "ToolInvocationError")
        if self._content:
            attribute = "gen_ai.tool.call.arguments" if key.endswith(".input") else "gen_ai.tool.call.result"
            # Tool results arrive as strings: write them as-is, so a JSON result stays a JSON object
            self._span.set_attribute(attribute, value if isinstance(value, str) else json.dumps(value, default=str))


class MapleHaystackTracer(OpenTelemetryTracer):
    def __init__(self, tracer: trace.Tracer, *, content: bool = True) -> None:
        super().__init__(tracer)
        self._content = content

    @contextmanager
    def trace(self, operation_name: str, tags: dict[str, Any] | None = None, parent_span: Any = None) -> Iterator[MapleSpan]:
        tags = dict(tags or {})
        attributes: dict[str, str] = {}
        operation = None
        if operation_name == "haystack.agent.run":
            # The Agent span has no name: use the pipeline component or the AgentTool that runs it
            parent = getattr(trace.get_current_span(), "attributes", None) or {}
            attributes["gen_ai.operation.name"] = "invoke_agent"
            attributes["gen_ai.agent.name"] = parent.get("haystack.component.name") or parent.get("gen_ai.tool.name") or "agent"
        elif operation_name == "haystack.agent.step.llm" or str(tags.get("haystack.component.type", "")).endswith("ChatGenerator"):
            operation = attributes["gen_ai.operation.name"] = "chat"
        elif operation_name == "haystack.agent.step.tool":
            operation = attributes["gen_ai.operation.name"] = "execute_tool"
            attributes["gen_ai.tool.name"] = tags["haystack.tool.name"]
        if conversation_id := _conversation_id.get():
            attributes["gen_ai.conversation.id"] = conversation_id
        if not self._content:
            tags.pop("haystack.pipeline.input_data", None)  # a plain tag, not gated by Haystack's content switch

        with self._tracer.start_as_current_span(operation_name, attributes=attributes) as raw_span:
            span = MapleSpan(raw_span, operation, self._content)
            span.set_tags(tags)
            yield span

Init once at startup, before the first pipeline.run()/agent.run() (no import-order constraint beyond that):

from haystack import tracing
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

from maple_haystack import MapleHaystackTracer

provider = TracerProvider(
    resource=Resource.create({"service.name": "<service>", "deployment.environment.name": "<env>"})
)
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(provider)
tracing.enable_tracing(MapleHaystackTracer(trace.get_tracer("haystack")))

Env:

OTEL_EXPORTER_OTLP_ENDPOINT=https://ingest.maple.dev   # EU: https://ingest.eu.maple.dev; /v1/traces is appended
OTEL_EXPORTER_OTLP_HEADERS="Authorization=Bearer <key>"
OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf
  • The app loads .env (load_dotenv(), --env-file): call it before OTLPSpanExporter() is built. Otherwise the exporter silently targets localhost:4318 with no key.
  • Tracer name must be "haystack" (instrumentation scope = how Maple labels the vendor Haystack).
  • Set a real service.name; never leave unknown_service.

Step 3: Session id

Haystack has no conversation concept; the app owns the message history. Wrap every run in conversation(<stable chat id>):

from maple_haystack import conversation

with conversation(chat_id):
    result = pipeline.run({"assistant": {"messages": [*history, ChatMessage.from_user(text)]}})
history = [m for m in result["assistant"]["messages"] if not m.is_from("system")]  # Agent re-adds its system prompt
  • chat_id = the id the app already stores for the chat/thread (DB id, frontend thread id). Never a per-request UUID, never a process-wide constant.
  • Wrap the run call itself (inside the request handler). It is a ContextVar: request-scoped and carried into Haystack's tool threads.
  • It writes gen_ai.conversation.id on every span. That is the only session key Maple reads for Haystack (session.id is ignored).

Step 4: Content

  • Default MapleHaystackTracer(..., content=True) writes gen_ai.input.messages, gen_ai.output.messages, gen_ai.system_instructions, gen_ai.tool.call.arguments/.result, plus Haystack's own haystack.* content tags.
  • content=False keeps model, tokens, cost, finish reason, tool names and failures; drops messages, tool args/results, haystack.* content tags and the ungated haystack.pipeline.input_data tag. Use it if the user asks for no prompts/PII in telemetry. It also replaces a failed tool's status message, which quotes the tool's arguments, with a generic one.
  • HAYSTACK_CONTENT_TRACING_ENABLED is irrelevant with this tracer; don't add it.

Step 5: Tools, errors, sub-agents

  • Tool spans: haystack.agent.step.tool → execute_tool + gen_ai.tool.name. A tool that raises becomes {"error": ...} in Haystack; the tracer sets status Error + error.type=ToolInvocationError. Don't catch exceptions inside tools just to return strings; let them raise (or return {"error": ...}).
  • Agent name (gen_ai.agent.name, needed for Maple lanes) = pipeline component name, or the AgentTool name, else agent. If the app calls agent.run() directly for its main agent, prefer running it as a Pipeline component with a meaningful name (only if that's a small change; otherwise leave it and mention the agent name).
  • Sub-agents: use AgentTool(agent=..., name="<worker>", description=...) (Haystack ≥3.1) or PipelineTool; each worker gets its own name/lane. Keep them under the same conversation() block.
  • Streaming with OpenAIChatGenerator: add generation_kwargs={"stream_options": {"include_usage": True}} or the streamed turn has no tokens. OpenRouter always sends usage.
  • Token mapping reads OpenAI-style meta["usage"] keys (prompt_tokens, completion_tokens, *_details). For another generator, print one reply's meta["usage"]; if keys differ, extend _chat().
  • Cost: only OpenRouter reports usage.cost → gen_ai.usage.cost. Other providers show "unpriced" in Maple; say so to the user.

Step 6: Flush

Scripts, CLIs, notebooks, jobs, serverless: flush before exit.

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

Servers: provider.shutdown() in the shutdown hook (FastAPI lifespan, atexit).

Step 7: Verify

Run one real conversation (≥2 turns, one tool call). No scriptable entry point (server, REPL, UI only) → write a small driver: one conversation() id, 2+ turns, at least one tool call, flush before exit. Without Maple access, both must hold: the run exits with no export errors on stderr (Failed to export, 401 lines), AND a local console/in-memory exporter shows the spans below. Silence alone proves nothing (no spans is silent too). With the Maple MCP: list_agent_sessions with search=<conversation id> returns one row. Check:

  • Every span has service.name set; spans' scope is haystack.
  • Each haystack.agent.step.llm span has gen_ai.operation.name=chat, gen_ai.response.model, gen_ai.usage.input_tokens, gen_ai.usage.output_tokens (also on the streamed turn).
  • gen_ai.input.messages/gen_ai.output.messages are JSON arrays of {role, parts} (content on).
  • Each tool call has one haystack.agent.step.tool span with the real gen_ai.tool.name; a failing tool has status Error and error.type; successful tools don't.
  • haystack.agent.run has gen_ai.operation.name=invoke_agent and a distinct gen_ai.agent.name per sub-agent.
  • Every trace of one conversation carries the same gen_ai.conversation.id; a second conversation carries a different one.
  • No model call appears twice (no second instrumentor).
  • No attribute contains an API key or Bearer .
  • In Maple: one session per conversation labelled Haystack, turns = runs, transcript non-empty, LLM/tool counts match, failed tool counted, cost shown (OpenRouter) or unpriced.

Tell the user: known gaps are no gen_ai.provider.name, no gen_ai.response.id, no tool call id on tool spans, and cost only on OpenRouter.

Expected trace

haystack.pipeline.run
  haystack.component.run            assistant
    haystack.agent.run              invoke_agent, agent "assistant"
      haystack.agent.step
        haystack.agent.step.llm     chat, openai/gpt-4o-mini
        haystack.agent.step.tool    execute_tool, get_weather
      haystack.agent.step
        haystack.agent.step.llm     chat

Known behaviours (expected; explain if the user asks)

  • Haystack looks up the active tracer on every span, so there is no import-order trap for enable_tracing(). Without this tracer, HAYSTACK_CONTENT_TRACING_ENABLED is read once at the first import haystack; setting it later silently records nothing.
  • Tools of one step run in parallel threads; their spans sit side by side under the step.
  • Haystack wraps a raised tool exception in ToolInvocationError, feeds the text back to the model, writes {"error": "..."} as the tool output and leaves the span Unset. The tracer sets Error regardless of raise_on_tool_invocation_failure.
  • An Agent inside a PipelineTool is named after its component in the inner pipeline.
  • Approval gates (ConfirmationHook at before_tool) stay in the same run and session. A rejected call never reaches the tool, so it has no tool span; the model sees the rejection as a tool result in the transcript. An Agent with hooks also gets a haystack.agent.hook span before its tool calls; like haystack.agent.step, it carries no model or tool attributes and isn't counted.
  • haystack.pipeline.input_data on the root span is a plain tag Haystack's own content switch never gated; content=False drops it. To redact rather than drop message content, filter values in _messages().

Do not

  • Don't set maple_ai.session.id for Haystack; gen_ai.conversation.id via conversation() is the key.