log_streams

​

LogStream

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

​

enable_metrics

def enable_metrics(self,
                   metrics: builtins.list[Union[GalileoScorers, Metric, LocalMetricConfig, str]]) -> builtins.list[LocalMetricConfig]

• metrics (builtins.list[Union[GalileoScorers, Metric, LocalMetricConfig, str]]): List of metrics to enable on this log stream. Supports multiple input formats:
  • GalileoScorers enum values: Built-in metrics like GalileoScorers.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”

• 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

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

from galileo.log_streams import LogStreams
from galileo.schema.metrics import GalileoScorers

# 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([
    GalileoScorers.correctness,
    GalileoScorers.completeness,
    "context_relevance",
    "toxicity"
])

logger.info(f"Server-side metrics enabled automatically")
logger.info(f"Need to process {len(local_metrics)} local 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([
    GalileoScorers.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}")

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

• 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: Optional[str]=None,
           project_name: Optional[str]=None) -> LogStream

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

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

• LogStream: The created log stream.

​

enable_metrics

def enable_metrics(self,
                   *,
                   log_stream_name: Optional[str]=None,
                   project_name: Optional[str]=None,
                   metrics: builtins.list[Union[GalileoScorers, Metric, LocalMetricConfig, str]]) -> builtins.list[LocalMetricConfig]

• 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[GalileoScorers, Metric, LocalMetricConfig, str]]): List of metrics to enable. Can include:
  • GalileoScorers enum values (e.g., GalileoScorers.correctness)
  • Metric objects with name and optional version
  • LocalMetricConfig objects for custom local metrics
  • String names of built-in metrics

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

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

# Enable built-in metrics with explicit parameters
from galileo.log_streams import LogStreams
from galileo.schema.metrics import GalileoScorers

log_streams = LogStreams()
scorer_configs, local_metrics = log_streams.enable_metrics(
    log_stream_name="Production Logs",
    project_name="My AI Project",
    metrics=[
        GalileoScorers.correctness,
        GalileoScorers.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: Optional[str]=None,
        name: Optional[str]=None,
        project_id: Optional[str]=None,
        project_name: Optional[str]=None) -> Optional[LogStream]

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

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

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

​

list

def list(self,
         *,
         project_id: Optional[str]=None,
         project_name: Optional[str]=None) -> builtins.list[LogStream]

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

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

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

​

create_log_stream

def create_log_stream(name: str,
                      project_id: Optional[str]=None,
                      project_name: Optional[str]=None) -> LogStream

• name (str): The name of the log stream.

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

• LogStream: The created project.

​

enable_metrics

def enable_metrics(*,
                   log_stream_name: Optional[str]=None,
                   project_name: Optional[str]=None,
                   metrics: builtins.list[Union[GalileoScorers, Metric, LocalMetricConfig, str]]) -> builtins.list[LocalMetricConfig]

• 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)

• 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[GalileoScorers, Metric, LocalMetricConfig, str]]): List of metrics to enable on the log stream. Can include:
  • GalileoScorers enum values (e.g., GalileoScorers.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”)

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

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

# Enable built-in metrics with explicit parameters
from galileo.log_streams import enable_metrics
from galileo.schema.metrics import GalileoScorers

    local_metrics = enable_metrics(
        log_stream_name="Production Logs",
        project_name="My AI Project",
        metrics=[
            GalileoScorers.correctness,
            GalileoScorers.completeness,
            "context_relevance",
            "instruction_adherence"
        ]
    )
    logger.info("Server-side metrics enabled automatically")
    logger.info(f"Need to process {len(local_metrics)} local metrics")

# 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", "toxicity"]
    )
    logger.info(f"Need to process {len(local_metrics)} local metrics")

# 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

# export GALILEO_PROJECT="My AI Project"
    local_metrics = enable_metrics(
        log_stream_name="Development Logs",  # Explicit log stream name
        # project_name automatically loaded from GALILEO_PROJECT env var
        metrics=[
            # Built-in metrics
            GalileoScorers.correctness,
            "toxicity",

            # Custom metric with specific version
            Metric(name="my_custom_metric", version=2),

            # Local custom metric
            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"Process local metric: {local_metric.name}")

​

get_log_stream

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

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

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

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

​

list_log_streams

def list_log_streams(*,
                     project_id: Optional[str]=None,
                     project_name: Optional[str]=None) -> list[LogStream]

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

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

• list[LogStream]: A list of Log streams.