log_stream

LogStream

Object-centric interface for Galileo log streams.

This class provides an intuitive way to work with Galileo log streams, offering methods for managing log streams and their associated metrics.

Examples

# Create a new log stream and persist it
log_stream = LogStream(name="Production Logs", project_name="My AI Project").create()

# Get an existing log stream
log_stream = LogStream.get(name="Production Logs", project_name="My AI Project")

# LogStreams can also be created through Project instances
from galileo.project import Project

project = Project.get(name="My AI Project")
log_stream = project.create_log_stream(name="Production Logs")

# Enable metrics on the log stream
from galileo.schema.metrics import GalileoMetrics
local_metrics = log_stream.enable_metrics([
    GalileoMetrics.correctness,
    GalileoMetrics.completeness,
    "context_relevance"
])

# Refresh log stream state from API
log_stream.refresh()

context

def context(self) -> Any

Get a galileo context manager for this log stream.

This is a convenient method that returns a pre-configured galileo_context for this log stream, eliminating the need to specify project and log stream names.

Examples

log_stream = LogStream.get(
    name="Production Logs",
    project_name="My AI Project"
)

with log_stream.context():
    # Your logging code here
    response = openai_client.chat.completions.create(...)

create

def create(self) -> LogStream

Persist this log stream to the API.

Examples

log_stream = LogStream(name="Production Logs", project_name="My AI Project").create()
assert log_stream.is_synced()

export_records

def export_records(self,
                   record_type: RecordType=RecordType.TRACE,
                   filters: builtins.list[FilterType] | None=None,
                   sort: LogRecordsSortClause=LogRecordsSortClause(column_id='created_at', ascending=False),
                   export_format: LLMExportFormat=LLMExportFormat.JSONL,
                   column_ids: builtins.list[str] | None=None,
                   redact: bool=True) -> Iterator[dict[str, Any]]

Export records from this log stream.

This method provides a convenient way to export records without needing to specify project_id or log_stream_id.

Arguments

• record_type: The type of records to export (SPAN, TRACE, or SESSION).
• filters: A list of filters to apply to the export.
• sort: A sort clause to order the exported records.
• export_format: The desired format for the exported data.
• column_ids: A list of column IDs to include in the export.
• redact: Redact sensitive data from the response.

get

def get(cls,
        *,
        name: str,
        project_id: str | None=None,
        project_name: str | None=None) -> LogStream | None

Get an existing log stream by name.

Arguments

• name (str): The log stream name.
• project_id (Optional[str]): The project ID. If neither project_id nor project_name is provided, falls back to GALILEO_PROJECT_ID or GALILEO_PROJECT environment variables.
• project_name (Optional[str]): The project name. If neither project_id nor project_name is provided, falls back to GALILEO_PROJECT environment variable.

get_metrics

def get_metrics(self) -> builtins.list[str]

Get the list of metrics currently enabled on this log stream.

Examples

log_stream = LogStream.get(name="Production Logs", project_name="My Project")
current_metrics = log_stream.get_metrics()
print(f"Currently enabled: {current_metrics}")

get_sessions

def get_sessions(self,
                 filters: builtins.list[FilterType] | None=None,
                 sort: LogRecordsSortClause | None=None,
                 limit: int=100,
                 starting_token: int=0) -> QueryResult

Query sessions in this log stream.

This is a convenience method that queries for sessions specifically.

Arguments

• filters: A list of filters to apply to the query.
• sort: A sort clause to order the query results.
• limit: The maximum number of records to return.
• starting_token: The token for the next page of results.

get_spans

def get_spans(self,
              filters: builtins.list[FilterType] | None=None,
              sort: LogRecordsSortClause | None=None,
              limit: int=100,
              starting_token: int=0) -> QueryResult

Query spans in this log stream.

This is a convenience method that queries for spans specifically.

Arguments

• filters: A list of filters to apply to the query.
• sort: A sort clause to order the query results.
• limit: The maximum number of records to return.
• starting_token: The token for the next page of results.

get_traces

def get_traces(self,
               filters: builtins.list[FilterType] | None=None,
               sort: LogRecordsSortClause | None=None,
               limit: int=100,
               starting_token: int=0) -> QueryResult

Query traces in this log stream.

This is a convenience method that queries for traces specifically.

Arguments

• filters: A list of filters to apply to the query.
• sort: A sort clause to order the query results.
• limit: The maximum number of records to return.
• starting_token: The token for the next page of results.

list

def list(cls,
         *,
         project_id: str | None=None,
         project_name: str | None=None,
         limit: Unset | int=100,
         starting_token: Unset | int=0) -> list[LogStream]

List log streams for a project.

Returns a single page of results. Use starting_token (from next_starting_token on a prior response) to fetch subsequent pages.

Arguments

• project_id (Optional[str]): The project ID. If neither project_id nor project_name is provided, falls back to GALILEO_PROJECT_ID or GALILEO_PROJECT environment variables.
• project_name (Optional[str]): The project name. If neither project_id nor project_name is provided, falls back to GALILEO_PROJECT environment variable.
• limit (Union[Unset, int]): Maximum number of log streams to return per page. Defaults to 100.
• starting_token (Union[Unset, int]): Pagination token to start from. Defaults to 0 (first page).

project

def project(self) -> Project | None

Get the project this log stream belongs to.

query

def query(self,
          record_type: RecordType,
          filters: builtins.list[FilterType] | None=None,
          sort: LogRecordsSortClause | None=None,
          limit: int=100,
          starting_token: int=0) -> QueryResult

Query records in this log stream.

This method provides a convenient way to search spans, traces, or sessions within the current log stream without needing to specify project_id or log_stream_id.

Arguments

• record_type: The type of records to query (SPAN, TRACE, or SESSION).
• filters: A list of filters to apply to the query.
• sort: A sort clause to order the query results.
• limit: The maximum number of records to return.
• starting_token: The token for the next page of results.

refresh

def refresh(self) -> None

Refresh this log stream’s state from the API.

Updates all attributes with the latest values from the remote API and sets the state to SYNCED.

Examples

log_stream.refresh()
assert log_stream.is_synced()

session_columns

def session_columns(self) -> ColumnCollection

Get available columns for sessions in this log stream.

Examples

log_stream = LogStream.get(name="Production Logs", project_name="My AI Project")
columns = log_stream.session_columns

# Access a specific column
model_column = columns["model"]

# Filter using columns
sessions = log_stream.get_sessions(
    filters=[columns["model"].equals("gpt-4o-mini")],
    sort=columns["created_at"].descending()
)

set_metrics

def set_metrics(self,
                metrics: builtins.list[GalileoMetrics | Metric | LocalMetricConfig | str]) -> builtins.list[LocalMetricConfig]

Set (replace) the metrics on this log stream.

This replaces any existing metrics with the new list. Alias for enable_metrics with clearer naming intent.

Arguments

• metrics: List of metrics to set. Supports:
  • GalileoMetrics enum values (e.g., GalileoMetrics.correctness)
  • Metric objects (including from Metric.get(id=”…”))
  • LocalMetricConfig objects for custom scoring functions
  • String names of built-in metrics

span_columns

def span_columns(self) -> ColumnCollection

Get available columns for spans in this log stream.

Examples

log_stream = LogStream.get(name="Production Logs", project_name="My AI Project")
columns = log_stream.span_columns

# Access a specific column
input_column = columns["input"]

# Filter using columns
spans = log_stream.get_spans(
    filters=[columns["input"].contains("world")],
    sort=columns["created_at"].descending()
)

trace_columns

def trace_columns(self) -> ColumnCollection

Get available columns for traces in this log stream.

Examples

log_stream = LogStream.get(name="Production Logs", project_name="My AI Project")
columns = log_stream.trace_columns

# Access a specific column
input_column = columns["input"]

# Filter using columns
traces = log_stream.get_traces(
    filters=[columns["input"].contains("largest")],
    sort=columns["created_at"].descending()
)