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

## LogStream

Log streams are used to organize logs within a project on the Galileo platform.

They provide a way to categorize and group related logs, making it easier to
analyze and monitor specific parts of your application or different environments
(e.g., production, staging, development).

**Arguments**

* `created_at` (`datetime.datetime`): The timestamp when the log stream was created.

* `created_by` (`str`): The identifier of the user who created the log stream.

* `id` (`str`): The unique identifier of the log stream.

* `name` (`str`): The name of the log stream.

* `project_id` (`str`): The ID of the project this log stream belongs to.

* `updated_at` (`datetime.datetime`): The timestamp when the log stream was last updated.

* `additional_properties` (`dict`): Additional properties associated with the log stream.

**Examples**

```python theme={null}
# Create a new log stream in a project
from galileo.log_streams import create_log_stream

# Create by project ID
log_stream = create_log_stream(name="Production Logs", project_id="project-123")

# Create by project name
log_stream = create_log_stream(name="Production Logs", project_name="My AI Project")

# Get a log stream by name
from galileo.log_streams import get_log_stream
log_stream = get_log_stream(name="Production Logs", project_name="My AI Project")

# List all log streams in a project
from galileo.log_streams import list_log_streams
log_streams = list_log_streams(project_name="My AI Project")
for stream in log_streams:
    logger.info(f"Log Stream: {stream.name} (ID: {stream.id})")

# Use a log stream with the context manager
from galileo.openai import openai
from galileo import galileo_context

with galileo_context(project="My AI Project", log_stream="Production Logs"):
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello, world!"}]
    )

# Enable metrics on a log stream - RECOMMENDED APPROACH
from galileo.log_streams import enable_metrics
from galileo.schema.metrics import GalileoMetrics

# Set environment variables first
# export GALILEO_LOG_STREAM="Production Logs"
# export GALILEO_PROJECT="My AI Project"

# Clean and simple - just pass the metrics!
local_metrics = enable_metrics([
    GalileoMetrics.correctness,
    GalileoMetrics.completeness,
    "context_relevance"
])

# Alternative: Use explicit parameters
local_metrics = enable_metrics(
    log_stream_name="Production Logs",
    project_name="My AI Project",
    metrics=["correctness", "completeness"]
)
```

### enable\_metrics

```python theme={null}
def enable_metrics(self,
                   metrics: builtins.list[Union[GalileoMetrics, Metric, LocalMetricConfig, str]]) -> builtins.list[LocalMetricConfig]
```

Enable metrics directly on this log stream instance.

This is the most intuitive and clean way to enable metrics when you already have a
LogStream object. The method leverages the log stream's existing project\_id and id
attributes, eliminating the need for redundant parameter specification and reducing
the potential for errors.

This approach is ideal for object-oriented workflows where you're working with LogStream
instances directly, and it provides the clearest semantic meaning: "enable these metrics
on this specific log stream."

**Arguments**

* `metrics` (`builtins.list[Union[GalileoMetrics, Metric, LocalMetricConfig, str]]`): List of metrics to enable on this log stream. Supports multiple input formats:

  * **GalileoMetrics enum values**: Built-in metrics like `GalileoMetrics.correctness`
  * **Metric objects**: Custom metrics with optional version specifications
  * **LocalMetricConfig objects**: Client-side metrics with custom scoring functions
  * **String names**: Built-in metric names like "correctness" or "toxicity"

**Raises**

* `ValueError`: - If this LogStream instance lacks required `id` or `project_id` attributes
* If any specified metrics are unknown or unavailable
* If there are issues with metric configuration or registration
* `GalileoHTTPException`: If there are network or API errors when communicating with Galileo services

**Returns**

* `builtins.list[LocalMetricConfig]`: List of local metric configurations that must be computed client-side.
  Server-side metrics are automatically registered with Galileo and don't
  need to be returned since users don't interact with them.

**Examples**

Basic usage with built-in metrics:

```python theme={null}
from galileo.log_streams import LogStreams
from galileo.schema.metrics import GalileoMetrics

# Get a log stream first
log_streams = LogStreams()
log_stream = log_streams.get(name="Production Logs", project_name="My AI Project")

# Enable metrics directly - clean and intuitive!
local_metrics = log_stream.enable_metrics([
    GalileoMetrics.correctness,
    GalileoMetrics.completeness,
    "context_relevance",
    "toxicity"
])

logger.info(f"Server-side metrics enabled automatically")
logger.info(f"Need to process {len(local_metrics)} local metrics")
```

Advanced usage with custom metrics:

```python theme={null}
from galileo.schema.metrics import Metric, LocalMetricConfig

def custom_scorer(trace_or_span):
    return 0.75  # Your scoring logic

local_metrics = log_stream.enable_metrics([
    GalileoMetrics.correctness,
    "completeness",
    Metric(name="domain_relevance", version=3),
    LocalMetricConfig(name="custom_metric", scorer_fn=custom_scorer)
])

# Process local metrics if any
for local_metric in local_metrics:
    logger.info(f"Need to process local metric: {local_metric.name}")
```

**Notes**

**Requirements:**

* The LogStream instance must have valid `id` and `project_id` attributes
* These are automatically set when retrieving LogStream objects via LogStreams methods

**Recommended Usage:**

* Use this method when you already have a LogStream object
* More intuitive than specifying project/log stream names again
* Cleaner object-oriented design pattern

## LogStreams

### create

```python theme={null}
def create(self,
           name: str,
           *,
           project_id: Optional[str]=None,
           project_name: Optional[str]=None) -> LogStream
```

Creates a new log stream. Exactly one of `project_id` or `project_name` must be provided.

**Arguments**

* `name` (`str`): The name of the log stream.
* `project_id` (`Optional[str]`): The ID of the project to create the log stream in. Defaults to None.
* `project_name` (`Optional[str]`): The name of the project to create the log stream in. Defaults to None.

**Raises**

* `ValueError`: If neither or both `project_id` and `project_name` are provided, or if the project is not found.
* `HTTPValidationError`: If the server validation fails.
* `errors.UnexpectedStatus`: If the server returns an undocumented status code and Client.raise\_on\_unexpected\_status is True.
* `httpx.TimeoutException`: If the request takes longer than Client.timeout.

**Returns**

* `LogStream`: The created log stream.

### enable\_metrics

```python theme={null}
def enable_metrics(self,
                   *,
                   log_stream_name: Optional[str]=None,
                   project_name: Optional[str]=None,
                   metrics: builtins.list[Union[GalileoMetrics, Metric, LocalMetricConfig, str]]) -> builtins.list[LocalMetricConfig]
```

Enable metrics for a log stream by configuring scorers.

The project name can be provided via the 'project\_name' parameter or the
GALILEO\_PROJECT environment variable.

The log stream name can be provided via the 'log\_stream\_name' parameter or the
GALILEO\_LOG\_STREAM environment variable.

**Arguments**

* `log_stream_name` (`Optional[str]`): The name of the log stream. Takes precedence over the GALILEO\_LOG\_STREAM environment variable. Defaults to None.
* `project_name` (`Optional[str]`): The name of the project. Takes precedence over the GALILEO\_PROJECT environment variable. Defaults to None.
* `metrics` (`builtins.list[Union[GalileoMetrics, Metric, LocalMetricConfig, str]]`): List of metrics to enable. Can include:
  * GalileoMetrics enum values (e.g., GalileoMetrics.correctness)
  * Metric objects with name and optional version
  * LocalMetricConfig objects for custom local metrics
  * String names of built-in metrics

**Raises**

* `ValueError`: If log stream or project cannot be found, or if metrics are unknown.

**Returns**

* `tuple[builtins.list[ScorerConfig], builtins.list[LocalMetricConfig]]`: A tuple containing the configured scorer configs and local metric configs.

**Examples**

```python theme={null}
# Enable built-in metrics with explicit parameters
from galileo.log_streams import LogStreams
from galileo.schema.metrics import GalileoMetrics

log_streams = LogStreams()
scorer_configs, local_metrics = log_streams.enable_metrics(
    log_stream_name="Production Logs",
    project_name="My AI Project",
    metrics=[
        GalileoMetrics.correctness,
        GalileoMetrics.completeness,
        "context_relevance",
    ],
)

# Enable metrics using environment variables
# export GALILEO_LOG_STREAM="Production Logs"
# export GALILEO_PROJECT="My AI Project"
scorer_configs, local_metrics = log_streams.enable_metrics(
    metrics=["correctness", "completeness"]
)

# Enable custom metrics with mixed parameters
from galileo.schema.metrics import Metric, LocalMetricConfig

def custom_scorer(trace_or_span):
    return 0.85  # Custom scoring logic

# export GALILEO_PROJECT="My AI Project"
scorer_configs, local_metrics = log_streams.enable_metrics(
    log_stream_name="Production Logs",  # Explicit log stream
    # project_name from env var
    metrics=[
        Metric(name="my_custom_metric", version=2),
        LocalMetricConfig(name="local_scorer", scorer_fn=custom_scorer)
    ]
)
```

### get

```python theme={null}
def get(self,
        *,
        id: Optional[str]=None,
        name: Optional[str]=None,
        project_id: Optional[str]=None,
        project_name: Optional[str]=None) -> Optional[LogStream]
```

Retrieves a log stream by id or name.

**Arguments**

* `id` (`Optional[str]`): The id of the log stream. Defaults to None.
* `name` (`Optional[str]`): The name of the log stream. Defaults to None.
* `project_id` (`Optional[str]`): The ID of the project. Defaults to None.
* `project_name` (`Optional[str]`): The name of the project. Defaults to None.

**Raises**

* `ValueError`: If neither or both `id` and `name` are provided, or if neither or both
  `project_id` and `project_name` are provided.
* `errors.UnexpectedStatus`: If the server returns an undocumented status code and Client.raise\_on\_unexpected\_status is True.
* `httpx.TimeoutException`: If the request takes longer than Client.timeout.

**Returns**

* `Optional[LogStream]`: The log stream if found, None otherwise.

### list

```python theme={null}
def list(self,
         *,
         project_id: Optional[str]=None,
         project_name: Optional[str]=None) -> builtins.list[LogStream]
```

Lists all log streams. Exactly one of `project_id` or `project_name` must be provided.

**Arguments**

* `project_id` (`Optional[str]`): The ID of the project to list log streams for.
* `project_name` (`Optional[str]`): The name of the project to list log streams for.

**Raises**

* `ValueError`: If neither or both `project_id` and `project_name` are provided.
* `errors.UnexpectedStatus`: If the server returns an undocumented status code and Client.raise\_on\_unexpected\_status is True.
* `httpx.TimeoutException`: If the request takes longer than Client.timeout.

**Returns**

* `builtins.list[LogStream]`: A list of log streams.

## create\_log\_stream

```python theme={null}
def create_log_stream(name: str,
                      project_id: Optional[str]=None,
                      project_name: Optional[str]=None) -> LogStream
```

Creates a new log stream. Exactly one of `project_id` or `project_name` must be provided.

**Arguments**

* `name` (`str`): The name of the log stream.

**Raises**

* `errors.UnexpectedStatus`: If the server returns an undocumented status code and Client.raise\_on\_unexpected\_status is True.
* `httpx.TimeoutException`: If the request takes longer than Client.timeout.

**Returns**

* `LogStream`: The created project.

## enable\_metrics

```python theme={null}
def enable_metrics(*,
                   log_stream_name: Optional[str]=None,
                   project_name: Optional[str]=None,
                   metrics: builtins.list[Union[GalileoMetrics, Metric, LocalMetricConfig, str]]) -> builtins.list[LocalMetricConfig]
```

Enable metrics for a log stream with flexible parameter and environment variable support.

This unified function supports both explicit parameters and environment variable fallbacks,
making it perfect for all use cases - from production CI/CD pipelines to development testing.

**Flexible Usage Patterns:**

* **Environment-only**: Just pass metrics, names from env vars (production/CI)
* **Explicit parameters**: Specify project/log stream names directly (development)
* **Mixed approach**: Combine explicit params with environment fallbacks

## Environment Variables (Optional Fallbacks)

GALILEO\_PROJECT : str
The name of the Galileo project (used when project\_name not provided)
GALILEO\_LOG\_STREAM : str
The name of the log stream (used when log\_stream\_name not provided)

**Arguments**

* `log_stream_name` (`Optional[str]`): The name of the log stream. Takes precedence over GALILEO\_LOG\_STREAM environment variable.
  If None, will use GALILEO\_LOG\_STREAM env var. Defaults to None.
* `project_name` (`Optional[str]`): The name of the project. Takes precedence over GALILEO\_PROJECT environment variable.
  If None, will use GALILEO\_PROJECT env var. Defaults to None.
* `metrics` (`builtins.list[Union[GalileoMetrics, Metric, LocalMetricConfig, str]]`): List of metrics to enable on the log stream. Can include:
  * GalileoMetrics enum values (e.g., GalileoMetrics.correctness)
  * Metric objects with name and optional version for custom metrics
  * LocalMetricConfig objects for client-side custom scoring functions
  * String names of built-in metrics (e.g., "correctness", "toxicity")

**Raises**

* `ValueError`: If log stream or project cannot be found, or if any specified metrics are unknown.
* `errors.UnexpectedStatus`: If the server returns an undocumented status code and Client.raise\_on\_unexpected\_status is True.
* `httpx.TimeoutException`: If the request takes longer than Client.timeout.

**Returns**

* `builtins.list[LocalMetricConfig]`: List of local metric configurations that must be computed client-side.
  Server-side metrics are automatically registered with Galileo and don't
  need to be returned since users don't interact with them.

**Examples**

```python theme={null}
# Enable built-in metrics with explicit parameters
from galileo.log_streams import enable_metrics
from galileo.schema.metrics import GalileoMetrics

local_metrics = enable_metrics(
    log_stream_name="Production Logs",
    project_name="My AI Project",
    metrics=[
        GalileoMetrics.correctness,
        GalileoMetrics.completeness,
        "context_relevance",
    ],
)

# Enable metrics using environment variables only
# export GALILEO_LOG_STREAM="Production Logs"
# export GALILEO_PROJECT="My AI Project"
local_metrics = enable_metrics(metrics=["correctness", "completeness"])

# Enable custom and local metrics with environment variable fallbacks
from galileo.schema.metrics import Metric, LocalMetricConfig
from galileo_core.schemas.logging.step import StepType

def response_length_scorer(trace_or_span):
    '''Custom metric that scores based on response length'''
    if hasattr(trace_or_span, "output") and trace_or_span.output:
        return min(len(trace_or_span.output) / 100.0, 1.0)  # Normalize 0-1
    return 0.0

local_metrics = enable_metrics(
    log_stream_name="Development Logs",
    metrics=[
        GalileoMetrics.correctness,
        "toxicity",
        Metric(name="my_custom_metric", version=2),
        LocalMetricConfig(
            name="response_length",
            scorer_fn=response_length_scorer,
            scorable_types=[StepType.llm],
            aggregatable_types=[StepType.trace],
        ),
    ],
)

# Process local metrics
for local_metric in local_metrics:
    logger.info(f"Need to process local metric: {local_metric.name}")
```

## get\_log\_stream

```python theme={null}
def get_log_stream(*,
                   name: Optional[str]=None,
                   project_id: Optional[str]=None,
                   project_name: Optional[str]=None) -> Optional[LogStream]
```

Retrieves a log stream by name. Exactly one of `project_id` or `project_name` must be provided.

**Arguments**

* `name` (`Optional[str]`): The name of the log stream. Defaults to None.
* `project_id` (`Optional[str]`): The ID of the project. Defaults to None.
* `project_name` (`Optional[str]`): The name of the project. Defaults to None.

**Raises**

* `ValueError`: If neither or both `project_id` and `project_name` are provided.
* `errors.UnexpectedStatus`: If the server returns an undocumented status code and Client.raise\_on\_unexpected\_status is True.
* `httpx.TimeoutException`: If the request takes longer than Client.timeout.

**Returns**

* `Optional[LogStream]`: The log stream if found, None otherwise.

## list\_log\_streams

```python theme={null}
def list_log_streams(*,
                     project_id: Optional[str]=None,
                     project_name: Optional[str]=None) -> list[LogStream]
```

Lists all log streams. Exactly one of `project_id` or `project_name` must be provided.

**Arguments**

* `project_id` (`str`): The id of the project.
* `project_name` (`str`): The name of the project.

**Raises**

* `errors.UnexpectedStatus`: If the server returns an undocumented status code and Client.raise\_on\_unexpected\_status is True.
* `httpx.TimeoutException`: If the request takes longer than Client.timeout.

**Returns**

* `list[LogStream]`: A list of Log streams.