decorator - Galileo

Module

Galileo Decorator Module. This module provides decorators for logging and tracing function calls in your application. Decorators allow you to add logging functionality to your existing code with minimal changes. How to use decorators:

Basic usage - decorate any function to log its execution:

from galileo import log

@log
def my_function(arg1, arg2):
    return arg1 + arg2

You can annotate your logs to help categorize and search logs later on:
```
@log(span_type="llm", name="GPT-4 Call")
def call_llm(prompt, temperature=0.7):
    # LLM call implementation
    return response
```
In this case, we are using:
1. span_type - This groups the logs here into the “llm” category.
2. name - This assigns a searchable name to all logs within this function.

Using context manager for grouping related operations:

from galileo import galileo_context

with galileo_context(project="my-project", log_stream="production"):
    result1 = my_function()
    result2 = another_function()

Setup requirements:

Galileo API key must be set (via environment variable GALILEO_API_KEY or programmatically)
Project and Log Stream names should be defined if using the log decorator (either via environment variables GALILEO_PROJECT and GALILEO_LOG_STREAM, or via galileo_context.init())

For more examples and detailed usage, see the Galileo SDK documentation.

GalileoDecorator

Main decorator class that provides both decorator and context manager functionality for logging and tracing in Galileo. This class can be used as:

A function decorator via the log method
A context manager via the __call__ method

clear_session

def clear_session(self) -> None

Clear the session in the active context logger instance.

flush

def flush(self,
          project: str | None=None,
          log_stream: str | None=None,
          experiment_id: str | None=None,
          mode: str | None=None,
          on_error: Callable[[Exception], None] | None=None) -> None

Upload all captured traces under a project and log stream context to Galileo. If no project or log stream is provided, then the currently initialized context is used. Arguments

project: The project name. Defaults to None.
log_stream: The log stream name. Defaults to None.
experiment_id: The experiment ID. Defaults to None.
mode: The logger mode. Defaults to None.
on_error: Optional callback invoked with the exception when a flush error occurs. If None, a warning is logged instead. Defaults to None.

flush_all

def flush_all(self) -> None

Upload all captured traces under all contexts to Galileo. This method flushes all traces regardless of project or log stream.

get_current_log_stream

def get_current_log_stream(self) -> str | None

Retrieve the current log stream name from context. Returns

str | None: The current log stream context

get_current_mode

def get_current_mode(self) -> LoggerModeType | None

Retrieve the current mode from context. Returns

Optional[LoggerModeType]: The current mode context, or None if not initialized

get_current_project

def get_current_project(self) -> str | None

Retrieve the current project name from context. Returns

str | None: The current project context

get_current_span_stack

def get_current_span_stack(self) -> list[WorkflowSpan]

Retrieve the current span stack from context. Returns

List[WorkflowSpan]: The current span stack

get_current_trace

def get_current_trace(self) -> Trace | None

Retrieve the current trace from context. Returns

Trace | None: The current trace

get_logger_instance

def get_logger_instance(self,
                        project: str | None=None,
                        log_stream: str | None=None,
                        experiment_id: str | None=None,
                        mode: str | None=None,
                        ingestion_hook: Callable | None=None) -> GalileoLogger

Get the Galileo Logger instance for the current decorator context. Arguments

project: Optional project name to use
log_stream: Optional log stream name to use
experiment_id: Optional experiment ID to use
mode: Optional logger mode to use

Returns

GalileoLogger instance configured with the specified project and log stream:

init

def init(self,
         project: str | None=None,
         log_stream: str | None=None,
         experiment_id: str | None=None,
         local_metrics: list[LocalMetricConfig] | None=None,
         mode: str | None=None) -> None

Initialize the context with a project and log stream. Optionally, it can also be used to start a trace. This method resets the existing active context with a new context with the specified project and log stream. Arguments

project: The project name. Defaults to None.
log_stream: The log stream name. Defaults to None.
experiment_id: The experiment id. Defaults to None.
local_metrics: Local metrics configs to run on the traces/spans before submitting them for ingestion. Defaults to None.
mode: The logger mode.

log

def log(self,
        func: Callable[P, R] | None=None,
        *,
        name: str | None=None,
        span_type: SPAN_TYPE | None=None,
        params: dict[str, str | Callable] | None=None,
        dataset_record: DatasetRecord | None=None) -> Callable[[Callable[P, R]], Callable[P, R]]

Main decorator function for logging function calls. This decorator can be used with or without arguments:

@log
@log(name=“my_function”, span_type=“llm”)

Arguments

func: The function to decorate (when used without parentheses)
name: Optional custom name for the span (defaults to function name)
span_type: Optional span type (“llm”, “retriever”, “tool”, “workflow”, “agent”)
params: Optional parameter mapping for extracting specific values
dataset_record: Optional parameter for dataset values. This is used by the local experiment module to set the dataset fields on the trace/spans and not generally provided for logging to log streams.

Returns

A decorated function that logs its execution:

reset

def reset(self) -> None

Reset the entire context, which also deletes all traces that haven’t been flushed. This method clears all context variables and resets the logger singleton.

reset_trace_context

def reset_trace_context(self) -> None

Reset the trace context inside the decorator.

set_session

def set_session(self, session_id: str) -> None

Set the session in the active context logger instance. This is useful when you want to continue logging to an existing session. Arguments

session_id: The id of the session to set.

start_session

def start_session(self,
                  name: str | None=None,
                  previous_session_id: str | None=None,
                  external_id: str | None=None,
                  metadata: dict[str, str] | None=None) -> str

Start a session in the active context logger instance. Arguments

name: The name of the session. If not provided, a session name will be generated automatically.
previous_session_id: The id of the previous session. Defaults to None.
external_id: The external id of the session. Defaults to None.
metadata: User metadata for the session. Defaults to None.

Returns

str: The id of the newly created session.

galileo_dataset_context

def galileo_dataset_context(*,
                            dataset_input: str | None=None,
                            dataset_output: str | None=None,
                            dataset_metadata: dict[str, str] | None=None) -> Generator[None, None, None]

Context manager to set dataset/ground truth information for spans. Use this when working with third-party OTEL instrumentation (e.g., Microsoft Agent Framework, LangChain, etc.) where you don’t have direct control over span creation but need to attach ground truth data for scorers that require reference output. Arguments

dataset_input (str): The expected input from your dataset (ground truth input).
dataset_output (str): The expected output/ground truth for scoring (e.g., for BLEU, exact match scorers).
dataset_metadata (dict[str, str]): Additional metadata from your dataset.

Examples

from galileo.decorator import galileo_dataset_context
>>>
# Set ground truth for a single agent call
with galileo_dataset_context(
    dataset_input="What is the capital of France?",
    dataset_output="Paris",
    dataset_metadata={"source": "geography_quiz"}
):
    response = await agent.run("What is the capital of France?")
# Use with experiment datasets
for row in dataset:
    with galileo_dataset_context(
        dataset_input=row["input"],
        dataset_output=row["expected_output"],
    ):
        response = await agent.run(row["input"])

Documentation Index

​Module

​GalileoDecorator

​clear_session

​flush

​flush_all

​get_current_log_stream

​get_current_mode

​get_current_project

​get_current_span_stack

​get_current_trace

​get_logger_instance

​init

​log

​reset

​reset_trace_context

​set_session

​start_session

​galileo_dataset_context

Module

GalileoDecorator

clear_session

flush

flush_all

get_current_log_stream

get_current_mode

get_current_project

get_current_span_stack

get_current_trace

get_logger_instance

init

log

reset

reset_trace_context

set_session

start_session

galileo_dataset_context