Overview
The Sessions Resource module provides synchronous and asynchronous clients for logging session 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 -- Sessions (synchronous) and AsyncSessions (asynchronous) -- that wrap the Phoenix v1/session_annotations REST API endpoint using httpx. Both classes provide three methods for logging annotations:
add_session_annotation() creates a single annotation for a session. It accepts a session_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 session. The method delegates to log_session_annotations() internally.
log_session_annotations() logs multiple session annotations in a single HTTP POST request. It accepts an iterable of SessionAnnotationData objects (TypedDict with session_id, name, annotator_kind, and result fields).
log_session_annotations_dataframe() ingests annotations from a pandas DataFrame, supporting both per-row and global annotation_name and annotator_kind values. The DataFrame's session_id can come from either a column or the index. Data is processed in chunks of 100 rows for efficient batch processing.
All methods support a sync parameter: when True, the request is processed synchronously and returns InsertedSessionAnnotation objects containing the inserted annotation IDs; when False (default), the request is processed asynchronously on the server and returns None.
The module re-exports the generated types InsertedSessionAnnotation, SessionAnnotationData, AnnotateSessionsRequestBody, and AnnotateSessionsResponseBody from v1.
Usage
Access the Sessions resource via the Phoenix client as client.sessions. Use it to add quality annotations to conversation sessions -- for example, marking sessions as helpful or toxic, logging LLM-judge evaluations, or bulk-importing annotation results from a DataFrame.
Code Reference
Source Location
Signature
class Sessions:
def __init__(self, client: httpx.Client) -> None: ...
def add_session_annotation(
self,
*,
session_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[InsertedSessionAnnotation]: ...
def log_session_annotations(
self,
*,
session_annotations: Iterable[SessionAnnotationData],
sync: bool = False,
) -> Optional[list[InsertedSessionAnnotation]]: ...
def log_session_annotations_dataframe(
self,
*,
dataframe: pd.DataFrame,
annotation_name: Optional[str] = None,
annotator_kind: Optional[Literal["LLM", "CODE", "HUMAN"]] = None,
sync: bool = False,
) -> Optional[list[InsertedSessionAnnotation]]: ...
class AsyncSessions:
def __init__(self, client: httpx.AsyncClient) -> None: ...
# Same methods as Sessions but async
Import
from phoenix.client.resources.sessions import Sessions, AsyncSessions
I/O Contract
add_session_annotation()
Inputs
| Name |
Type |
Required |
Description
|
| session_id |
str |
Yes |
The ID of the session 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 session
|
| sync |
bool |
No |
If True, returns the inserted annotation; if False (default), returns None
|
Outputs
| Name |
Type |
Description
|
| return |
Optional[InsertedSessionAnnotation] |
Inserted annotation with ID if sync=True; None if sync=False
|
log_session_annotations()
Inputs
| Name |
Type |
Required |
Description
|
| session_annotations |
Iterable[SessionAnnotationData] |
Yes |
Iterable of annotation data dicts with session_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[InsertedSessionAnnotation]] |
List of inserted annotations if sync=True; None if sync=False
|
log_session_annotations_dataframe()
Inputs
| Name |
Type |
Required |
Description
|
| dataframe |
pd.DataFrame |
Yes |
DataFrame with session annotation data; session_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[InsertedSessionAnnotation]] |
List of all inserted annotations if sync=True; None if sync=False
|
Usage Examples
from phoenix.client import Client
client = Client()
# Add a single session annotation
annotation = client.sessions.add_session_annotation(
session_id="session_abc123",
annotation_name="helpfulness",
annotator_kind="HUMAN",
label="helpful",
score=0.9,
explanation="The session provided accurate and clear responses.",
sync=True,
)
# Log multiple annotations at once
annotations = [
{
"session_id": "session_abc123",
"name": "toxicity",
"annotator_kind": "LLM",
"result": {"label": "safe", "score": 0.05},
},
{
"session_id": "session_def456",
"name": "toxicity",
"annotator_kind": "LLM",
"result": {"label": "safe", "score": 0.02},
},
]
client.sessions.log_session_annotations(session_annotations=annotations)
# Log annotations from a DataFrame
import pandas as pd
df = pd.DataFrame({
"session_id": ["session_abc123", "session_def456"],
"label": ["positive", "negative"],
"score": [0.9, 0.3],
})
client.sessions.log_session_annotations_dataframe(
dataframe=df,
annotation_name="sentiment",
annotator_kind="LLM",
)
Related Pages