Documentation Index
Fetch the complete documentation index at: https://docs.galileo.ai/llms.txt
Use this file to discover all available pages before exploring further.
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
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
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]]
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
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]
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
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
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
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]
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
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
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
log_stream.refresh()
assert log_stream.is_synced()
session_columns
def session_columns(self) -> ColumnCollection
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]
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
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
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()
)
