Manual instrumentation guide

┌─────────────────────────────────────────────────────────┐
│  Your Agent Application (Python)                        │
│                                                         │
│  ┌──────────────────────────────────────────────────┐   │
│  │ opentelemetry-instrument CLI wrapper             │   │
│  │ (auto-configures TracerProvider, LoggerProvider, │   │
│  │  sets up OTLP exporters)                         │   │
│  └──────────────────────────────────────────────────┘   │
│                                                         │
│  ┌─────────────────────────────────────────────────┐    │
│  │ Your code (manual instrumentation):             │    │
│  │  - _tracer.start_as_current_span(...)  → spans  │    │
│  │  - _otel_logger.emit(...)              → logs   │    │
│  └─────────────────────────────────────────────────┘    │
│                    │                                    │
│                    │ OTLP (protobuf)                    │
│                    ▼                                    │
│  ┌─────────────────────────────────────────────────┐    │
│  │ BatchSpanProcessor + BatchLogRecordProcessor    │    │
│  │ (buffers, batches, exports periodically)        │    │
│  └─────────────────────────────────────────────────┘    │
└────────────────────│────────────────────────────────────┘
                     │ OTLP HTTP (:4318) or gRPC (:4317)
                     ▼
┌─────────────────────────────────────────────────────────┐
│  OTel Collector                                         │
│  - Receives traces + logs via OTLP                      │
│  - Batches and exports to backend(s)                    │
│  - Routes traces and logs to separate streams           │
└────────────────────│────────────────────────────────────┘
                     │ OTLP/HTTP (JSON)
                     ▼
┌─────────────────────────────────────────────────────────┐
│  Parseable                                              │
│  - genai-traces stream (flattened spans)                │
│  - genai-logs stream (flattened log records)            │
│  - SQL-queryable, with server-side cost enrichment      │
└─────────────────────────────────────────────────────────┘

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk._logs import LoggerProvider
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
from opentelemetry._logs import set_logger_provider

# Traces
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(tracer_provider)

# Logs
logger_provider = LoggerProvider()
logger_provider.add_log_record_processor(BatchLogRecordProcessor(OTLPLogExporter()))
set_logger_provider(logger_provider)

pip install opentelemetry-distro opentelemetry-exporter-otlp
opentelemetry-bootstrap -a install

# Required: where the OTel Collector listens
export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318

# Required: OTLP protocol
export OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf

# Required: identifies your application in traces
export OTEL_SERVICE_NAME=my-genai-agent

# Required: disable auto-instrumentor to prevent duplicates
export OTEL_PYTHON_DISABLED_INSTRUMENTATIONS=openai_v2

opentelemetry-instrument \
    --traces_exporter otlp \
    --logs_exporter otlp \
    --metrics_exporter none \
    python my_agent.py

from opentelemetry import trace
from opentelemetry._logs import get_logger_provider, SeverityNumber

_tracer = trace.get_tracer("my-agent", "1.1.0")
_otel_logger = get_logger_provider().get_logger("my-agent", "1.1.0")

import json
from opentelemetry import trace
from opentelemetry._logs import get_logger_provider, SeverityNumber

_tracer = trace.get_tracer("my-agent.llm", "1.1.0")
_otel_logger = get_logger_provider().get_logger("my-agent.llm", "1.1.0")

def run_agent(model_name: str, provider: str, problem: str):
    """Example: wrap an agent run with an invoke_agent span."""

    span_name = "invoke_agent my-agent"
    with _tracer.start_as_current_span(span_name, kind=trace.SpanKind.CLIENT) as span:
        span.set_attribute("gen_ai.operation.name", "invoke_agent")
        span.set_attribute("gen_ai.agent.name", "my-agent")
        span.set_attribute("gen_ai.provider.name", provider)
        span.set_attribute("gen_ai.request.model", model_name)

        # ── Emit the problem statement that triggered this agent run ──
        _otel_logger.emit(
            body=problem,
            severity_number=SeverityNumber.INFO,
            event_name="gen_ai.user.message",
            attributes={
                "gen_ai.operation.name": "invoke_agent",
                "gen_ai.provider.name": provider,
                "gen_ai.request.model": model_name,
                "gen_ai.agent.name": "my-agent",
                "gen_ai.event.name": "gen_ai.user.message",
                "role": "user",
            },
        )

        # ── Your agent loop ──
        done = False
        while not done:
            llm_response = call_llm(model_name, messages)    # creates a child "chat" span
            tool_result = execute_tool(llm_response.action)   # creates a child "execute_tool" span
            done = llm_response.is_done

        # ── After loop: set aggregate response attributes ──
        span.set_attribute("gen_ai.usage.input_tokens", total_input_tokens)
        span.set_attribute("gen_ai.usage.output_tokens", total_output_tokens)
        span.set_attribute("gen_ai.response.finish_reasons", json.dumps(["exit_command"]))

        # ── Emit agent completion summary ──
        _otel_logger.emit(
            body=json.dumps({"exit_status": "exit_command",
                             "total_input_tokens": total_input_tokens,
                             "total_output_tokens": total_output_tokens}),
            severity_number=SeverityNumber.INFO,
            event_name="gen_ai.agent.finish",
            attributes={
                "gen_ai.operation.name": "invoke_agent",
                "gen_ai.agent.name": "my-agent",
                "gen_ai.provider.name": provider,
                "gen_ai.request.model": model_name,
                "gen_ai.event.name": "gen_ai.agent.finish",
                "gen_ai.usage.input_tokens": total_input_tokens,
                "gen_ai.usage.output_tokens": total_output_tokens,
            },
        )

def execute_tool(action: str, tool_call_id: str | None = None):
    """Example: wrap a tool execution with an execute_tool span."""

    tool_name = action.strip().split()[0] if action.strip() else "unknown"
    span_name = f"execute_tool {tool_name}"
    with _tracer.start_as_current_span(span_name, kind=trace.SpanKind.INTERNAL) as span:
        span.set_attribute("gen_ai.operation.name", "execute_tool")
        span.set_attribute("gen_ai.tool.name", tool_name)
        span.set_attribute("gen_ai.tool.type", "function")
        if tool_call_id:
            span.set_attribute("gen_ai.tool.call.id", tool_call_id)

        # Emit tool input log (the command being run)
        _otel_logger.emit(
            body=action,
            severity_number=SeverityNumber.INFO,
            event_name="gen_ai.tool.input",
            attributes={
                "gen_ai.operation.name": "execute_tool",
                "gen_ai.tool.name": tool_name,
                "gen_ai.tool.type": "function",
                "gen_ai.tool.call.id": tool_call_id or "",
                "gen_ai.event.name": "gen_ai.tool.input",
            },
        )

        try:
            result = env.communicate(action)
        except TimeoutError as e:
            span.set_attribute("error.type", type(e).__name__)
            span.set_status(trace.StatusCode.ERROR, "Command timed out")
            raise

        # Emit tool output log (the observation)
        _otel_logger.emit(
            body=result or "",
            severity_number=SeverityNumber.INFO,
            event_name="gen_ai.tool.output",
            attributes={
                "gen_ai.operation.name": "execute_tool",
                "gen_ai.tool.name": tool_name,
                "gen_ai.tool.type": "function",
                "gen_ai.tool.call.id": tool_call_id or "",
                "gen_ai.event.name": "gen_ai.tool.output",
            },
        )

        return result

def call_llm(model: str, messages: list[dict], temperature: float = 0.0, **kwargs):
    """Example: instrument any LLM call with OTel traces + logs."""

    with _tracer.start_as_current_span(f"chat {model}", kind=trace.SpanKind.CLIENT) as span:

        # ── Step 1: Set request attributes on span ──
        span.set_attribute("gen_ai.operation.name", "chat")
        span.set_attribute("gen_ai.request.model", model)
        span.set_attribute("gen_ai.provider.name", "openai")  # or "anthropic", etc.
        if temperature is not None:
            span.set_attribute("gen_ai.request.temperature", temperature)

        # ── Step 2: Emit input message log records ──
        for msg in messages:
            role = msg.get("role", "user")
            content = msg.get("content", "")
            body = content if isinstance(content, str) else json.dumps(content)
            _otel_logger.emit(
                body=body,
                severity_number=SeverityNumber.INFO,
                event_name=f"gen_ai.{role}.message",
                attributes={
                    "gen_ai.provider.name": "openai",
                    "gen_ai.request.model": model,
                    "gen_ai.event.name": f"gen_ai.{role}.message",
                    "role": role,
                },
            )

        # ── Step 3: Call the LLM ──
        try:
            response = your_llm_client.chat.completions.create(
                model=model,
                messages=messages,
                temperature=temperature,
                **kwargs,
            )
        except Exception as e:
            span.set_status(trace.StatusCode.ERROR, str(e))
            span.set_attribute("error.type", type(e).__name__)
            raise

        # ── Step 4: Set response attributes on span ──
        if response.model:
            span.set_attribute("gen_ai.response.model", response.model)
        if response.id:
            span.set_attribute("gen_ai.response.id", response.id)
        if response.usage:
            span.set_attribute("gen_ai.usage.input_tokens", response.usage.prompt_tokens)
            span.set_attribute("gen_ai.usage.output_tokens", response.usage.completion_tokens)

        # ── Step 5: Emit response log records ──
        finish_reasons = []
        for i, choice in enumerate(response.choices):
            # Choice content
            _otel_logger.emit(
                body=choice.message.content or "",
                severity_number=SeverityNumber.INFO,
                event_name="gen_ai.choice",
                attributes={
                    "gen_ai.provider.name": "openai",
                    "gen_ai.request.model": model,
                    "gen_ai.event.name": "gen_ai.choice",
                    "index": i,
                    "finish_reason": choice.finish_reason or "",
                },
            )
            if choice.finish_reason:
                finish_reasons.append(choice.finish_reason)

            # Tool calls (if present)
            if choice.message.tool_calls:
                tool_call_ids = []
                for tc in choice.message.tool_calls:
                    tc_dict = tc.model_dump() if hasattr(tc, "model_dump") else tc
                    if tc_dict.get("id"):
                        tool_call_ids.append(tc_dict["id"])
                    _otel_logger.emit(
                        body=json.dumps(tc_dict.get("function", tc_dict)),
                        severity_number=SeverityNumber.INFO,
                        event_name="gen_ai.tool.call",
                        attributes={
                            "gen_ai.provider.name": "openai",
                            "gen_ai.request.model": model,
                            "gen_ai.event.name": "gen_ai.tool.call",
                            "gen_ai.tool.name": tc_dict.get("function", {}).get("name", ""),
                            "gen_ai.tool.call.id": tc_dict.get("id", ""),
                        },
                    )

                # Set tool call IDs on the chat span for cross-span correlation
                # with execute_tool spans. Single ID stored as string, multiple
                # IDs JSON-encoded as array.
                if tool_call_ids:
                    span.set_attribute(
                        "gen_ai.tool.call.id",
                        tool_call_ids[0] if len(tool_call_ids) == 1 else json.dumps(tool_call_ids),
                    )

            # Thinking/reasoning blocks (Claude, DeepSeek, etc.)
            thinking_blocks = getattr(choice.message, "thinking_blocks", None)
            if thinking_blocks:
                for tb in thinking_blocks:
                    thinking_text = tb.get("thinking", "") if isinstance(tb, dict) else str(tb)
                    _otel_logger.emit(
                        body=thinking_text,
                        severity_number=SeverityNumber.INFO,
                        event_name="gen_ai.thinking",
                        attributes={
                            "gen_ai.provider.name": "openai",
                            "gen_ai.request.model": model,
                            "gen_ai.event.name": "gen_ai.thinking",
                        },
                    )

        # ── Step 6: Finalize span ──
        if finish_reasons:
            span.set_attribute("gen_ai.response.finish_reasons", json.dumps(finish_reasons))
        span.set_status(trace.StatusCode.OK)

        return response

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  batch:
    timeout: 5s
    send_batch_size: 100

exporters:
  # Traces -> Parseable
  otlphttp/traces:
    endpoint: https://your-parseable-instance:8000
    encoding: json
    headers:
      Authorization: "Basic <credentials>"
      X-P-Stream: "genai-traces"
      X-P-Log-Source: "otel-traces"

  # Logs -> Parseable
  otlphttp/logs:
    endpoint: https://your-parseable-instance:8000
    encoding: json
    headers:
      Authorization: "Basic <credentials>"
      X-P-Stream: "genai-logs"
      X-P-Log-Source: "otel-logs"

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [batch]
      exporters: [otlphttp/traces]
    logs:
      receivers: [otlp]
      processors: [batch]
      exporters: [otlphttp/logs]

docker run -d --name otel-collector \
  -p 4317:4317 -p 4318:4318 \
  -v $(pwd)/otel-collector-config.yaml:/etc/otelcol/config.yaml \
  otel/opentelemetry-collector-contrib:latest

otelcol-contrib --config otel-collector-config.yaml

response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is 2+2?"},
    ],
)

messages = [
    {"role": "system", "content": "You are a math tutor."},
    {"role": "user", "content": "What is 10*5?"},
    {"role": "assistant", "content": "50"},
    {"role": "user", "content": "Divide by 2"},
]
response = call_llm("gpt-4o", messages)

try:
    response = call_llm(...)
except RateLimitError as e:
    # The span still exists with ERROR status
    pass

with _tracer.start_as_current_span(f"chat {model}", kind=trace.SpanKind.CLIENT) as span:
    span.set_attribute("gen_ai.request.model", model)
    # ... emit input message logs ...

    stream = client.chat.completions.create(model=model, messages=messages, stream=True)
    chunks = []
    for chunk in stream:
        chunks.append(chunk)

    # Reconstruct full response from chunks, then emit logs
    full_response_text = "".join(c.choices[0].delta.content or "" for c in chunks if c.choices)
    _otel_logger.emit(body=full_response_text, event_name="gen_ai.choice", ...)

    # Token usage may not be available in streaming mode (provider-dependent)
    span.set_status(trace.StatusCode.OK)

for attempt in range(max_retries):
    try:
        response = call_llm(model, messages)  # Each call creates its own span
        break
    except RateLimitError:
        time.sleep(backoff)

_tracer = trace.get_tracer("my-agent.llm", "1.0.0")
_otel_logger = get_logger_provider().get_logger("my-agent.llm", "1.0.0")

echo $OTEL_PYTHON_DISABLED_INSTRUMENTATIONS
# Should output: openai_v2