Phoenix — Testing, CI & Production¶

# tests/conftest.py
import os
import pytest
from openinference.instrumentation import using_attributes
from openinference.instrumentation.anthropic import AnthropicInstrumentor
from phoenix.client import Client
from phoenix.otel import register

PROJECT = os.getenv("PHOENIX_PROJECT_NAME", "triage-ci")
RUN_ID = os.getenv("CI_PIPELINE_ID", "local")

# SimpleSpanProcessor (batch=False): each span is exported when it ends
tracer_provider = register(project_name=PROJECT, protocol="http/protobuf", verbose=False)
AnthropicInstrumentor().instrument(tracer_provider=tracer_provider)


@pytest.fixture(scope="session")
def phoenix() -> Client:
    return Client()


@pytest.fixture(autouse=True)
def phoenix_session(request):
    """One Phoenix session per test: session.id = pytest node id."""
    session_id = f"{RUN_ID}:{request.node.nodeid}"
    with using_attributes(
        session_id=session_id,
        metadata={"test_id": request.node.nodeid, "run_id": RUN_ID, "git_sha": os.getenv("GIT_SHA", "")},
        tags=["pytest", *[m.name for m in request.node.iter_markers()]],
    ):
        yield session_id
    request.node.user_properties.append(("phoenix_session", session_id))   # lands in JUnit XML


def pytest_sessionfinish(session, exitstatus):
    tracer_provider.force_flush()

# tests/phoenix_helpers.py
import time

import httpx
from phoenix.client import Client


def wait_for_spans(px: Client, project: str, session_id: str, min_count: int = 1, timeout: float = 15.0) -> list[dict]:
    deadline = time.monotonic() + timeout
    while True:
        try:
            spans = px.spans.get_spans(
                project_identifier=project,
                attributes={"session.id": session_id},  # server >= 14.9
                limit=500,
            )
        except httpx.HTTPStatusError as exc:             # 404 until the project's first span lands
            if exc.response.status_code != 404:
                raise
            spans = []
        if len(spans) >= min_count or time.monotonic() > deadline:
            return spans
        time.sleep(0.5)


def by_kind(spans: list[dict], kind: str) -> list[dict]:
    return [s for s in spans if s["span_kind"] == kind]

# tests/test_triage_agent.py
from app.agent import triage_agent
from tests.conftest import PROJECT
from tests.phoenix_helpers import by_kind, wait_for_spans


def test_refund_ticket_uses_billing_tool_once(phoenix, phoenix_session):
    result = triage_agent("I was charged twice for March, please refund")
    assert result["queue"] == "billing"

    spans = wait_for_spans(phoenix, PROJECT, phoenix_session, min_count=3)
    tools = [s["attributes"]["tool.name"] for s in by_kind(spans, "TOOL")]
    llm_calls = by_kind(spans, "LLM")

    assert tools.count("lookup_invoices") == 1                   # called exactly once
    assert "issue_refund" not in tools                            # agent must not refund on its own
    assert len(llm_calls) <= 3                                    # no runaway loop
    assert all(s["status_code"] != "ERROR" for s in spans)
    total_tokens = sum(s["attributes"].get("llm.token_count.total", 0) for s in llm_calls)
    assert total_tokens < 4_000, f"token budget exceeded: {total_tokens}"

import pytest
from phoenix.client.pytest import evaluate, log_evaluation, log_output
from phoenix.evals import LLM
from phoenix.evals.metrics import CorrectnessEvaluator

from app.rag import answer

judge = LLM(provider="anthropic", model="claude-haiku-4-5")
correctness = CorrectnessEvaluator(llm=judge)

CASES = [
    ("How long is the refund window?", "30 days"),
    ("Can I export reports to CSV?", "Settings → Export"),
]


@pytest.mark.phoenix(dataset="docs-rag-regression", repetitions=2)
@pytest.mark.parametrize("question,must_mention", CASES, ids=["refund-window", "csv-export"])
def test_docs_answers(question, must_mention):
    reply = answer(question)
    log_output({"answer": reply})
    log_evaluation(name="length_ok", score=float(len(reply) < 600))          # recorded, does not fail
    result = evaluate(correctness, input=question, output=reply)            # recorded + returned
    assert must_mention.lower() in reply.lower()
    assert result[0].label == "correct"

# scripts/eval_gate.py — run in CI after unit tests
import sys
from phoenix.client import Client
from phoenix.client.experiments import run_experiment

from app.triage import route_ticket

THRESHOLDS = {"queue_match": 0.92, "valid_queue": 1.0}

px = Client()
dataset = px.datasets.get_dataset(dataset="triage-golden", splits=["smoke", "adversarial"])


def queue_match(output, expected) -> bool:
    return output["queue"] == expected["queue"]


def valid_queue(output) -> bool:
    return output["queue"] in {"billing", "bugs", "how-to", "security"}


exp = run_experiment(
    dataset=dataset,
    task=lambda input: route_ticket(input["ticket"]),
    evaluators=[queue_match, valid_queue],
    experiment_name=f"ci-{sys.argv[1]}",                 # commit SHA
    experiment_metadata={"sha": sys.argv[1], "branch": sys.argv[2]},
    client=px,
)

failed = []
for name, minimum in THRESHOLDS.items():
    scores = [r.result["score"] for r in exp["evaluation_runs"] if r.name == name and r.result]
    mean = sum(scores) / max(len(scores), 1)
    print(f"{name}: {mean:.3f} (min {minimum})")
    if mean < minimum:
        failed.append(name)

print(px.experiments.get_experiment_url(exp["dataset_id"], exp["experiment_id"]))
sys.exit(1 if failed else 0)

# .github/workflows/llm-eval.yml (fragment)
jobs:
  eval-gate:
    runs-on: ubuntu-latest
    services:
      phoenix:
        image: arizephoenix/phoenix:20.16.0      # pinned
        ports: ["6006:6006", "4317:4317"]
    env:
      PHOENIX_COLLECTOR_ENDPOINT: http://localhost:6006
      ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v6
      - run: uv sync
      - run: uv run python scripts/seed_datasets.py         # upload golden set from the repo
      - run: uv run python scripts/eval_gate.py "$GITHUB_SHA" "$GITHUB_REF_NAME"