pip install arize-phoenix-client

# For local Phoenix server (default)export PHOENIX_ENDPOINT="http://localhost:6006"# A hosted or self-hosted instance with authenticationexport PHOENIX_ENDPOINT="https://phoenix.example.com"export PHOENIX_API_KEY="your-api-key"# Customize headersexport PHOENIX_CLIENT_HEADERS="Authorization=Bearer your-api-key,custom-header=value"

from phoenix.client import Client, AsyncClient# Automatic configuration from environment variablesclient = Client()client = Client(base_url="http://localhost:6006")  # Local Phoenix server# Cloud instance with API keyclient = Client(    base_url="https://your-phoenix.example.com",    api_key="your-api-key")# Custom authentication headersclient = Client(    base_url="https://your-phoenix-instance.com",    headers={"Authorization": "Bearer your-api-key"})# Asynchronous client (same configuration options)async_client = AsyncClient()async_client = AsyncClient(base_url="http://localhost:6006")async_client = AsyncClient(    base_url="https://your-phoenix.example.com",    api_key="your-api-key")

from phoenix.client import Clientfrom phoenix.client.types import PromptVersioncontent = """You're an expert educator in {{ topic }}. Summarize the following articlein a few concise bullet points that are easy for beginners to understand.{{ article }}"""prompt = client.prompts.create(    name="article-bullet-summarizer",    version=PromptVersion(        [{"role": "user", "content": content}],        model_name="gpt-4o-mini",    ),    prompt_description="Summarize an article in a few bullet points")# Retrieve and use promptsprompt = client.prompts.get(prompt_identifier="article-bullet-summarizer")# Format the prompt with variablesprompt_vars = {    "topic": "Sports",    "article": "Moises Henriques, the Australian all-rounder, has signed to play for Surrey in this summer's NatWest T20 Blast. He will join after the IPL and is expected to strengthen the squad throughout the campaign."}formatted_prompt = prompt.format(variables=prompt_vars)# Make a request with your Prompt using OpenAIfrom openai import OpenAIoai_client = OpenAI()resp = oai_client.chat.completions.create(**formatted_prompt)print(resp.choices[0].message.content)

import pandas as pd# List all available datasetsdatasets = client.datasets.list()for dataset in datasets:    print(f"Dataset: {dataset['name']} ({dataset['example_count']} examples)")# Get a specific dataset with all examplesdataset = client.datasets.get_dataset(dataset="qa-evaluation")print(f"Dataset {dataset.name} has {len(dataset)} examples")# Convert dataset to pandas DataFrame for analysisdf = dataset.to_dataframe()print(df.columns)  # Index(['input', 'output', 'metadata'], dtype='object')# Create a new dataset from dictionariesdataset = client.datasets.create_dataset(    name="customer-support-qa",    dataset_description="Q&A dataset for customer support evaluation",    inputs=[        {"question": "How do I reset my password?"},        {"question": "What's your return policy?"},        {"question": "How do I track my order?"}    ],    outputs=[        {"answer": "You can reset your password by clicking the 'Forgot Password' link on the login page."},        {"answer": "We offer 30-day returns for unused items in original packaging."},        {"answer": "You can track your order using the tracking number sent to your email."}    ],    metadata=[        {"category": "account", "difficulty": "easy"},        {"category": "policy", "difficulty": "medium"},        {"category": "orders", "difficulty": "easy"}    ])# Create dataset from pandas DataFramedf = pd.DataFrame({    "prompt": ["Hello", "Hi there", "Good morning"],    "response": ["Hi! How can I help?", "Hello! What can I do for you?", "Good morning! How may I assist?"],    "sentiment": ["neutral", "positive", "positive"],    "length": [20, 25, 30]})dataset = client.datasets.create_dataset(    name="greeting-responses",    dataframe=df,    input_keys=["prompt"],           # Columns to use as input    output_keys=["response"],        # Columns to use as expected output    metadata_keys=["sentiment", "length"]  # Additional metadata columns)

from datetime import datetime, timedeltafrom phoenix.client.types.spans import SpanQuery# Get spans as pandas DataFrame for analysisspans_df = client.spans.get_spans_dataframe(    project_identifier="my-llm-app",    limit=1000,    query=SpanQuery().where("parent_id is None"),  # Only top-level spans    start_time=datetime.now() - timedelta(hours=24))# Get span annotations as DataFrameannotations_df = client.spans.get_span_annotations_dataframe(    spans_dataframe=spans_df,  # Use spans from previous query    project_identifier="my-llm-app",    include_annotation_names=["relevance", "accuracy"],  # Only specific annotations    exclude_annotation_names=["note"]  # Exclude UI notes)

# Add a single annotation with human feedbackclient.spans.add_span_annotation(    span_id="span-123",    annotation_name="helpfulness",    annotator_kind="HUMAN",    label="helpful",    score=0.9,    explanation="Response directly answered the user's question")# Bulk annotation logging for multiple spansannotations = [    {        "name": "sentiment",        "span_id": "span-123",        "annotator_kind": "LLM",        "result": {"label": "positive", "score": 0.8}    },    {        "name": "accuracy",        "span_id": "span-456",        "annotator_kind": "HUMAN",        "result": {"label": "accurate", "score": 0.95}    },]client.spans.log_span_annotations(span_annotations=annotations)

# List all projectsprojects = client.projects.list()for project in projects:    print(f"Project: {project['name']} (ID: {project['id']})")# Filter server-side by a case-insensitive substring of the project namesupport_projects = client.projects.list(name_contains="support")# Create a new projectnew_project = client.projects.create(    name="Customer Support Bot",    description="Traces and evaluations for our customer support chatbot")print(f"Created project with ID: {new_project['id']}")

# Get a specific session by IDsession = client.sessions.get(session_id="my-session-id")print(f"Session: {session['session_id']}, Traces: {len(session['traces'])}")# List sessions for a projectsessions = client.sessions.list(project_name="my-chatbot")for s in sessions:    print(f"{s['session_id']}: {len(s['traces'])} traces")# List with a limitrecent = client.sessions.list(project_name="my-chatbot", limit=10)# Get sessions as a pandas DataFramedf = client.sessions.get_sessions_dataframe(project_name="my-chatbot")print(df[["session_id", "start_time", "end_time", "num_traces"]])

from phoenix.client import AsyncClientasync_client = AsyncClient()session = await async_client.sessions.get(session_id="my-session-id")sessions = await async_client.sessions.list(project_name="my-chatbot", limit=10)df = await async_client.sessions.get_sessions_dataframe(project_name="my-chatbot")