> ## 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.

# 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**

```python theme={null}
# 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

```python theme={null}
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**

```python theme={null}
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

```python theme={null}
def create(self) -> LogStream
```

Persist this log stream to the API.

**Examples**

```python theme={null}
log_stream = LogStream(name="Production Logs", project_name="My AI Project").create()
assert log_stream.is_synced()
```

### export\_records

```python theme={null}
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

```python theme={null}
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

```python theme={null}
def get_metrics(self) -> builtins.list[str]
```

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

**Examples**

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
def project(self) -> Project | None
```

Get the project this log stream belongs to.

### query

```python theme={null}
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

```python theme={null}
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**

```python theme={null}
log_stream.refresh()
assert log_stream.is_synced()
```

### session\_columns

```python theme={null}
def session_columns(self) -> ColumnCollection
```

Get available columns for sessions in this log stream.

**Examples**

```python theme={null}
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

```python theme={null}
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

```python theme={null}
def span_columns(self) -> ColumnCollection
```

Get available columns for spans in this log stream.

**Examples**

```python theme={null}
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

```python theme={null}
def trace_columns(self) -> ColumnCollection
```

Get available columns for traces in this log stream.

**Examples**

```python theme={null}
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()
)
```