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

# 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

def enable_metrics(self,
                   metrics: builtins.list[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:

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:

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

def create(self,
           name: str,
           *,
           project_id: str | None=None,
           project_name: str | None=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

def enable_metrics(self,
                   *,
                   log_stream_name: str | None=None,
                   project_name: str | None=None,
                   metrics: builtins.list[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

# 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

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

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

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

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

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 ID of the project to list log streams for.
• project_name (Optional[str]): The name of the project to list log streams for.
• limit (Union[Unset, int]): The maximum number of log streams to return per page. Defaults to 100.
• starting_token (Union[Unset, int]): The pagination token to start from. Defaults to 0 (first page).

Raises

• ValueError: If neither or both project_id and project_name are provided, if the named project is not found, if the server returns a validation error, or if the response is unexpectedly empty.
• 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 page of log streams.

create_log_stream

def create_log_stream(name: str,
                      project_id: str | None=None,
                      project_name: str | None=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

def enable_metrics(*,
                   log_stream_name: str | None=None,
                   project_name: str | None=None,
                   metrics: builtins.list[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

# 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

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

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

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

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

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

Arguments

• project_id (str): The id of the project.
• project_name (str): The name of the project.
• limit (Union[Unset, int]): The maximum number of log streams to return per page. Defaults to 100.
• starting_token (Union[Unset, int]): The pagination token to start from. Defaults to 0 (first page).

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

• builtins.list[LogStream]: A page of log streams.