Skip to main content

​
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()
)