Implementation:Arize ai Phoenix Log Span Annotations

Knowledge Sources	Phoenix
Domains	AI Observability, Batch Processing, Span Evaluation
Last Updated	2026-02-14 00:00 GMT

Overview

Concrete tool for submitting multiple span annotations in bulk, provided by the arize-phoenix-client package, with support for both typed dictionaries and pandas DataFrames as input formats.

Description

The Spans.log_span_annotations() method accepts an iterable of SpanAnnotationData typed dictionaries and posts them to the v1/span_annotations REST endpoint in a single HTTP request. When sync=True, the query parameter ?sync=true is appended and the server processes the annotations synchronously, returning a list of InsertedSpanAnnotation objects with their assigned IDs.

The Spans.log_span_annotations_dataframe() method provides a DataFrame-oriented interface. It validates the DataFrame structure, then processes it in chunks of 100 rows. Each chunk is converted to a list of SpanAnnotationData dicts and delegated to log_span_annotations(). Global annotation_name and annotator_kind parameters can override or supplement per-row values.

Additionally, Spans.add_document_annotation() and Spans.log_document_annotations() provide the same single and batch patterns for annotating individual retrieved documents within a span (identified by span_id + document_position). These use the v1/document_annotations endpoint.

Usage

Use log_span_annotations() when:

• You have a pre-built list of annotation dictionaries from a programmatic evaluation pipeline.
• You need fine-grained control over the exact annotation data structure.

Use log_span_annotations_dataframe() when:

• Working in a data-science context where annotations are computed and stored in a pandas DataFrame.
• Importing evaluation results from a notebook or CSV file.
• Applying a uniform annotation name or annotator kind to all rows.

Use add_document_annotation() or log_document_annotations() when:

• Scoring individual documents within a retrieval span (e.g., relevance of each retrieved passage in a RAG pipeline).

Code Reference

Source Location

• Repository: Phoenix
• File: packages/phoenix-client/src/phoenix/client/resources/spans/__init__.py
• Lines: 901-960 (log_span_annotations), 776-900 (log_span_annotations_dataframe), 961-1080 (add_document_annotation), 1082-1170 (log_document_annotations)

Signature

def log_span_annotations(
    self,
    *,
    span_annotations: Iterable[SpanAnnotationData],
    sync: bool = False,
) -> Optional[list[InsertedSpanAnnotation]]:
    ...

def log_span_annotations_dataframe(
    self,
    *,
    dataframe: "pd.DataFrame",
    annotator_kind: Optional[Literal["LLM", "CODE", "HUMAN"]] = None,
    annotation_name: Optional[str] = None,
    sync: bool = False,
) -> Optional[list[InsertedSpanAnnotation]]:
    ...

def add_document_annotation(
    self,
    *,
    span_id: str,
    document_position: int,
    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,
    sync: bool = False,
) -> Optional[InsertedSpanDocumentAnnotation]:
    ...

def log_document_annotations(
    self,
    *,
    document_annotations: list[SpanDocumentAnnotationData],
    sync: bool = False,
) -> Optional[list[InsertedSpanDocumentAnnotation]]:
    ...

Import

from phoenix.client import Client

client = Client()
# Access via: client.spans.log_span_annotations(...)
# Access via: client.spans.log_span_annotations_dataframe(...)
# Access via: client.spans.add_document_annotation(...)
# Access via: client.spans.log_document_annotations(...)

I/O Contract

Inputs (log_span_annotations)

Name	Type	Required	Description
span_annotations	Iterable[SpanAnnotationData]	Yes	An iterable of annotation data dictionaries. Each dict must contain: span_id (str), name (str), annotator_kind (str: "HUMAN", "LLM", or "CODE"), and result (dict with optional label, score, explanation). Optional keys: metadata (dict), identifier (str). Must not be empty.
sync	bool	No	If True, the server processes annotations synchronously and returns inserted annotation IDs. If False (default), annotations are queued asynchronously.

Inputs (log_span_annotations_dataframe)

Name	Type	Required	Description
dataframe	pd.DataFrame	Yes	A pandas DataFrame containing annotation data. Must include a span_id column (or use the DataFrame index). Must include either a name or annotation_name column (but not both), or the global annotation_name parameter must be provided. Must include an annotator_kind column or the global annotator_kind parameter. Optional columns: label, score, explanation, metadata, identifier.
annotator_kind	Optional[Literal["LLM", "CODE", "HUMAN"]]	No	Global annotator kind applied to all rows. Overrides the DataFrame column if both are present.
annotation_name	Optional[str]	No	Global annotation name applied to all rows. Overrides the DataFrame column if both are present.
sync	bool	No	If True, returns inserted annotation IDs for all chunks. If False (default), returns None.

Outputs

Name	Type	Description
(log_span_annotations return)	Optional[list[InsertedSpanAnnotation]]	When sync=True, a list of dicts each containing the id of the inserted/updated annotation. When sync=False, None.
(log_span_annotations_dataframe return)	Optional[list[InsertedSpanAnnotation]]	When sync=True, a combined list of all inserted annotations across all chunks. When sync=False, None.

Usage Examples

Batch Annotate with Typed Dictionaries

from phoenix.client import Client

client = Client()

annotations = [
    {
        "span_id": "span_001",
        "name": "relevance",
        "annotator_kind": "LLM",
        "result": {
            "label": "relevant",
            "score": 0.92,
            "explanation": "Response directly answers the question.",
        },
    },
    {
        "span_id": "span_002",
        "name": "relevance",
        "annotator_kind": "LLM",
        "result": {
            "label": "irrelevant",
            "score": 0.15,
            "explanation": "Response discusses unrelated topics.",
        },
    },
]

result = client.spans.log_span_annotations(
    span_annotations=annotations,
    sync=True,
)
print(f"Inserted {len(result)} annotations")

Batch Annotate from a DataFrame

import pandas as pd
from phoenix.client import Client

client = Client()

df = pd.DataFrame({
    "span_id": ["span_001", "span_002", "span_003"],
    "label": ["good", "bad", "good"],
    "score": [0.9, 0.2, 0.85],
    "explanation": [
        "Accurate and helpful",
        "Contains factual errors",
        "Mostly correct with minor omissions",
    ],
})

client.spans.log_span_annotations_dataframe(
    dataframe=df,
    annotation_name="quality",
    annotator_kind="HUMAN",
)

Batch Annotate with Per-Row Names and Annotator Kinds

import pandas as pd
from phoenix.client import Client

client = Client()

df = pd.DataFrame({
    "span_id": ["span_001", "span_002"],
    "annotation_name": ["toxicity", "helpfulness"],
    "annotator_kind": ["CODE", "LLM"],
    "label": ["safe", "helpful"],
    "score": [0.0, 0.88],
})

inserted = client.spans.log_span_annotations_dataframe(
    dataframe=df,
    sync=True,
)
for anno in inserted:
    print(f"Annotation ID: {anno['id']}")

Annotate Individual Documents in a Retrieval Span

from phoenix.client import Client

client = Client()

# Score each retrieved document in a RAG retrieval span
for position in range(5):
    client.spans.add_document_annotation(
        span_id="retrieval_span_001",
        document_position=position,
        annotation_name="relevance",
        annotator_kind="LLM",
        score=0.9 - (position * 0.15),
        label="relevant" if position < 3 else "irrelevant",
    )

Related Pages

Implements Principle

• Principle:Arize_ai_Phoenix_Batch_Span_Annotation