Implementation:Arize ai Phoenix Traces Resource

Knowledge Sources	Arize_ai_Phoenix
Domains	AI_Observability, Client_SDK
Last Updated	2026-02-14 05:30 GMT

Overview

The Traces Resource module provides synchronous and asynchronous clients for logging trace annotations to the Phoenix REST API, with support for individual annotations, bulk annotation lists, and DataFrame-based batch ingestion.

Description

This module defines two classes -- Traces (synchronous) and AsyncTraces (asynchronous) -- that wrap the Phoenix v1/trace_annotations REST API endpoint using httpx. Both classes provide three methods for logging annotations:

add_trace_annotation() creates a single annotation for a trace. It accepts a trace_id, annotation_name, annotator_kind ("LLM", "CODE", or "HUMAN"), and optional label, score, explanation, metadata, and identifier fields. The identifier parameter enables multiple annotations with the same name on the same trace. Internally, it delegates to log_trace_annotations().

log_trace_annotations() logs multiple trace annotations in a single HTTP POST request. It accepts an iterable of TraceAnnotationData objects (TypedDict with trace_id, name, annotator_kind, and result fields). Raises ValueError if the iterable is empty.

log_trace_annotations_dataframe() ingests annotations from a pandas DataFrame, supporting both per-row and global annotation_name and annotator_kind values. The DataFrame's trace_id can come from either a column or the index. Data is processed in chunks of 100 rows for efficient batch processing. Validation is performed by _validate_trace_annotations_dataframe() before processing.

All methods support a sync parameter: when True, the request is processed synchronously and returns InsertedTraceAnnotation objects containing the inserted annotation IDs; when False (default), the request is processed asynchronously on the server and returns None. The overloaded type signatures ensure proper return type narrowing based on the sync literal value.

The module re-exports the generated types InsertedTraceAnnotation, TraceAnnotationData, AnnotateTracesRequestBody, and AnnotateTracesResponseBody from v1.

Usage

Access the Traces resource via the Phoenix client as client.traces. Use it to add quality annotations to traces -- for example, marking traces as correct or incorrect, logging LLM-judge evaluations on entire traces, or bulk-importing evaluation results from a DataFrame.

Code Reference

Source Location

Repository: Arize_ai_Phoenix
File: packages/phoenix-client/src/phoenix/client/resources/traces/__init__.py
Lines: 621

Signature

class Traces:
    def __init__(self, client: httpx.Client) -> None: ...
    def add_trace_annotation(
        self,
        *,
        trace_id: str,
        annotation_name: str,
        annotator_kind: Literal["LLM", "CODE", "HUMAN"] = "HUMAN",
        label: Optional[str] = None,
        score: Optional[float] = None,
        explanation: Optional[str] = None,
        metadata: Optional[dict[str, Any]] = None,
        identifier: Optional[str] = None,
        sync: bool = False,
    ) -> Optional[InsertedTraceAnnotation]: ...
    def log_trace_annotations(
        self,
        *,
        trace_annotations: Iterable[TraceAnnotationData],
        sync: bool = False,
    ) -> Optional[list[InsertedTraceAnnotation]]: ...
    def log_trace_annotations_dataframe(
        self,
        *,
        dataframe: pd.DataFrame,
        annotation_name: Optional[str] = None,
        annotator_kind: Optional[Literal["LLM", "CODE", "HUMAN"]] = None,
        sync: bool = False,
    ) -> Optional[list[InsertedTraceAnnotation]]: ...

class AsyncTraces:
    def __init__(self, client: httpx.AsyncClient) -> None: ...
    # Same methods as Traces but async

Import

from phoenix.client.resources.traces import Traces, AsyncTraces

I/O Contract

`add_trace_annotation()`

Inputs

Name	Type	Required	Description
trace_id	`str`	Yes	The ID of the trace to annotate
annotation_name	`str`	Yes	The name of the annotation
annotator_kind	`Literal["LLM", "CODE", "HUMAN"]`	No	Kind of annotator (default: "HUMAN")
label	`Optional[str]`	No	Label assigned by the annotation (at least one of label, score, or explanation required)
score	`Optional[float]`	No	Score assigned by the annotation
explanation	`Optional[str]`	No	Explanation of the annotation result
metadata	`Optional[dict[str, Any]]`	No	Additional metadata for the annotation
identifier	`Optional[str]`	No	Unique identifier enabling multiple annotations with same name on same trace
sync	`bool`	No	If True, returns the inserted annotation; if False (default), returns None

Outputs

Name	Type	Description
return	`Optional[InsertedTraceAnnotation]`	Inserted annotation with ID if sync=True; None if sync=False

`log_trace_annotations()`

Inputs

Name	Type	Required	Description
trace_annotations	`Iterable[TraceAnnotationData]`	Yes	Iterable of annotation data dicts with trace_id, name, annotator_kind, and result
sync	`bool`	No	If True, returns inserted annotations; if False (default), returns None

Outputs

Name	Type	Description
return	`Optional[list[InsertedTraceAnnotation]]`	List of inserted annotations if sync=True; None if sync=False

`log_trace_annotations_dataframe()`

Inputs

Name	Type	Required	Description
dataframe	`pd.DataFrame`	Yes	DataFrame with trace annotation data; trace_id from column or index
annotation_name	`Optional[str]`	No	Global annotation name applied to all rows (overrides DataFrame column)
annotator_kind	`Optional[Literal["LLM", "CODE", "HUMAN"]]`	No	Global annotator kind applied to all rows
sync	`bool`	No	If True, returns inserted annotations; if False (default), returns None

Outputs

Name	Type	Description
return	`Optional[list[InsertedTraceAnnotation]]`	List of all inserted annotations if sync=True; None if sync=False

Usage Examples

from phoenix.client import Client

client = Client()

# Add a single trace annotation with synchronous response
annotation = client.traces.add_trace_annotation(
    trace_id="abc123def456",
    annotation_name="correctness",
    annotator_kind="HUMAN",
    label="correct",
    score=0.95,
    explanation="The trace produces the correct answer with proper reasoning.",
    sync=True,
)
print(f"Inserted annotation ID: {annotation['id']}")

# Log multiple trace annotations in bulk
annotations = [
    {
        "trace_id": "abc123def456",
        "name": "toxicity",
        "annotator_kind": "LLM",
        "result": {"label": "safe", "score": 0.01},
    },
    {
        "trace_id": "ghi789jkl012",
        "name": "toxicity",
        "annotator_kind": "LLM",
        "result": {"label": "safe", "score": 0.03},
    },
]
client.traces.log_trace_annotations(trace_annotations=annotations)

# Log annotations from a DataFrame
import pandas as pd

df = pd.DataFrame({
    "trace_id": ["abc123def456", "ghi789jkl012"],
    "label": ["correct", "incorrect"],
    "score": [0.95, 0.2],
    "explanation": ["Good answer", "Wrong reasoning"],
})
results = client.traces.log_trace_annotations_dataframe(
    dataframe=df,
    annotation_name="correctness",
    annotator_kind="HUMAN",
    sync=True,
)

# Async usage
from phoenix.client import AsyncClient

async_client = AsyncClient()
await async_client.traces.add_trace_annotation(
    trace_id="abc123def456",
    annotation_name="relevance",
    score=0.8,
    sync=True,
)

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment