Phoenix — Tracing & Instrumentation¶

from phoenix.otel import register

tracer_provider = register(
    project_name="ticket-triage-dev",            # or PHOENIX_PROJECT_NAME
    endpoint="http://localhost:6006/v1/traces",  # or PHOENIX_COLLECTOR_ENDPOINT
    batch=True,                                  # BatchSpanProcessor (default: Simple)
    auto_instrument=True,                        # enable all installed openinference instrumentors
)
tracer = tracer_provider.get_tracer(__name__)

from openinference.instrumentation.anthropic import AnthropicInstrumentor
from openinference.instrumentation.litellm import LiteLLMInstrumentor

# Explicit instead of auto_instrument=True — predictable in tests
AnthropicInstrumentor().instrument(tracer_provider=tracer_provider)
LiteLLMInstrumentor().instrument(tracer_provider=tracer_provider)

import litellm
litellm.completion(
    model="anthropic/claude-sonnet-5",
    messages=[{"role": "user", "content": "Summarise ticket T-812 in one line"}],
)
# -> LLM span: llm.model_name, llm.input_messages.*, llm.output_messages.*, llm.token_count.*

from phoenix.otel import register

tracer = register(project_name="ticket-triage-dev").get_tracer("triage")


@tracer.tool
def lookup_customer(email: str) -> dict:
    """Fetch the customer plan from CRM."""
    return {"email": email, "plan": "pro", "open_tickets": 2}


@tracer.chain
def classify(ticket: str) -> str:
    customer = lookup_customer("ann@example.com")
    return "billing" if "charged" in ticket.lower() else "bugs"


@tracer.agent
def triage_agent(ticket: str) -> dict:
    queue = classify(ticket)
    return {"queue": queue, "priority": "high"}

from opentelemetry.trace import Status, StatusCode
from openinference.semconv.trace import DocumentAttributes, SpanAttributes


def retrieve(query: str, k: int = 3) -> list[dict]:
    with tracer.start_as_current_span("search_kb", openinference_span_kind="retriever") as span:
        span.set_input(query)
        docs = kb.search(query, k=k)                     # your vector store
        for i, doc in enumerate(docs):
            prefix = f"{SpanAttributes.RETRIEVAL_DOCUMENTS}.{i}"
            span.set_attribute(f"{prefix}.{DocumentAttributes.DOCUMENT_ID}", doc["id"])
            span.set_attribute(f"{prefix}.{DocumentAttributes.DOCUMENT_CONTENT}", doc["text"])
            span.set_attribute(f"{prefix}.{DocumentAttributes.DOCUMENT_SCORE}", doc["score"])
        span.set_status(Status(StatusCode.OK))
        return docs


def answer(question: str) -> str:
    with tracer.start_as_current_span("rag_answer", openinference_span_kind="chain") as span:
        span.set_input(question)
        docs = retrieve(question)
        reply = llm_answer(question, docs)                # instrumented SDK call -> LLM child span
        span.set_output(reply)
        return reply

from openinference.instrumentation import using_attributes, using_session, using_user


def handle_chat_turn(conversation_id: str, user_id: str, message: str) -> str:
    with using_attributes(
        session_id=conversation_id,              # groups turns into a session (chat view in UI)
        user_id=user_id,
        metadata={"app_version": "2.3.1", "channel": "web"},
        tags=["triage", "beta"],
    ):
        return triage_agent(message)

# otel-collector.yaml
exporters:
  otlphttp/phoenix:
    endpoint: http://phoenix:6006          # the exporter appends /v1/traces
    headers:
      authorization: "Bearer ${env:PHOENIX_API_KEY}"
processors:
  filter/llm-only:                          # keep only OpenInference spans
    traces:
      span:
        - 'attributes["openinference.span.kind"] == nil'
service:
  pipelines:
    traces/phoenix:
      receivers: [otlp]
      processors: [filter/llm-only, batch]
      exporters: [otlphttp/phoenix]

from opentelemetry import trace
from phoenix.client import Client

px = Client()


def triage_endpoint(ticket: str) -> dict:
    result = triage_agent(ticket)
    span_id = format(trace.get_current_span().get_span_context().span_id, "016x")
    return {**result, "span_id": span_id}          # return it to the frontend


def on_user_feedback(span_id: str, helpful: bool, comment: str | None) -> None:
    px.spans.add_span_annotation(
        span_id=span_id,
        annotation_name="user_feedback",
        annotator_kind="HUMAN",                    # HUMAN | LLM | CODE
        label="helpful" if helpful else "not_helpful",
        score=1.0 if helpful else 0.0,
        explanation=comment,
    )