LlamaIndex observability — traces missing, exporters not firing

# pip install arize-phoenix openinference-instrumentation-llama-index

# --- Initialize BEFORE building components ---
import phoenix as px

# Launch the local Phoenix UI at http://localhost:6006
px.launch_app()

# Set up the LlamaIndex instrumentation
from phoenix.otel import register
tracer_provider = register(
    project_name="my-app-dev",
    endpoint="http://localhost:6006/v1/traces",   # local Phoenix
)

from openinference.instrumentation.llama_index import LlamaIndexInstrumentor
LlamaIndexInstrumentor().instrument(tracer_provider=tracer_provider)

# --- NOW build components ---
from llama_index.core import VectorStoreIndex
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-4o-mini")
index = VectorStoreIndex.from_documents(docs, embed_model=embed_model)
query_engine = index.as_query_engine(llm=llm, similarity_top_k=6)

# Every LLM call, embedding call, and retrieval will now appear in Phoenix
response = query_engine.query("What is the refund policy?")

# Open http://localhost:6006 to see:
#   - Traces tree per query
#   - LLM inputs/outputs
#   - Retrieved node scores
#   - Timing per step
#   - Cost estimates

# pip install langfuse openinference-instrumentation-llama-index

import os
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..."
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..."
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com"   # or self-hosted

from langfuse import Langfuse
from langfuse.llama_index import LlamaIndexCallbackHandler

# Newer OpenInference-based path
from openinference.instrumentation.llama_index import LlamaIndexInstrumentor

# LlamaIndex will export via OTLP to Langfuse's OpenTelemetry endpoint
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

tracer_provider = TracerProvider()
otlp_exporter = OTLPSpanExporter(
    endpoint=f"{os.environ['LANGFUSE_HOST']}/api/public/otel/v1/traces",
    headers={"Authorization": f"Basic {base64.b64encode(f'{os.environ['LANGFUSE_PUBLIC_KEY']}:{os.environ['LANGFUSE_SECRET_KEY']}'.encode()).decode()}"},
)
tracer_provider.add_span_processor(BatchSpanProcessor(otlp_exporter))
trace.set_tracer_provider(tracer_provider)

LlamaIndexInstrumentor().instrument(tracer_provider=tracer_provider)

# --- Now use LlamaIndex normally ---
response = query_engine.query("What is the refund policy?")

# --- For serverless: flush before exit ---
def flush_traces():
    tracer_provider.force_flush(timeout_millis=5000)

import atexit
atexit.register(flush_traces)

# Traces appear in Langfuse UI with cost, latency, and full span tree

from llama_index.core.instrumentation import get_dispatcher
from llama_index.core.instrumentation.event_handlers import BaseEventHandler
from llama_index.core.instrumentation.events import (
    LLMChatStartEvent, LLMChatEndEvent,
    EmbeddingEndEvent,
    RetrievalStartEvent, RetrievalEndEvent,
)
import time
import json

class CustomLoggingHandler(BaseEventHandler):
    """Ship every event to your own logging / observability."""

    def handle(self, event) -> None:
        payload = {
            "timestamp": time.time(),
            "type": type(event).__name__,
            "id": event.id_,
        }
        if isinstance(event, LLMChatStartEvent):
            payload["messages"] = [str(m) for m in event.messages]
        elif isinstance(event, LLMChatEndEvent):
            payload["response"] = str(event.response) if event.response else None
        elif isinstance(event, RetrievalEndEvent):
            payload["num_nodes"] = len(event.nodes) if event.nodes else 0
            payload["top_score"] = event.nodes[0].score if event.nodes else None

        # Ship to your log aggregator / message queue
        my_logger.info(json.dumps(payload))

    @classmethod
    def class_name(cls) -> str:
        return "CustomLoggingHandler"

# --- Attach to the global dispatcher ---
dispatcher = get_dispatcher()
dispatcher.add_event_handler(CustomLoggingHandler())

# --- Now every LlamaIndex event flows to your handler ---
response = query_engine.query("...")

# --- For per-request scoped context, use span handlers ---
from llama_index.core.instrumentation.span_handlers import BaseSpanHandler

class RequestScopedSpanHandler(BaseSpanHandler):
    def new_span(self, id_: str, bound_args, instance, parent_span_id=None, **kwargs):
        # Attach request_id from a context var to every span
        # (use contextvars.ContextVar to thread through async)
        pass

    def prepare_to_exit_span(self, id_, bound_args, instance, result=None, **kwargs):
        pass

    def prepare_to_drop_span(self, id_, bound_args, instance, err=None, **kwargs):
        pass

    @classmethod
    def class_name(cls) -> str:
        return "RequestScopedSpanHandler"

dispatcher.add_span_handler(RequestScopedSpanHandler())