Langfuse Manager

Langfuse manager for dataset and experiment tracker.

LangfuseFields

Constants for Langfuse field names.

LangfuseTracesSearchParams

Constants for Langfuse search params.

add_dataset_to_langfuse_if_not_exists(langfuse_client, dataset_name, dataset, dataset_description='', metadata=None, is_append=False)

Add the dataset to Langfuse.

Parameters:

Name	Type	Description	Default
langfuse_client	Langfuse	The Langfuse client.	required
dataset_name	str	The name of the dataset in Langfuse.	required
dataset	list[dict[str, Any]]	The dataset to add to Langfuse.	required
dataset_description	str	The description of the dataset.	''
metadata	dict[str, Any] | None	The metadata of the dataset.	None
is_append	bool	If True, append items to existing dataset. If False, only create if dataset doesn't exist.	False

Returns:

Name	Type	Description
str	str	The name of the created dataset in Langfuse.

aggregate_scores_for_whole_session(traces_list)

Aggregate the scores for the whole session and calculate the average score, with engine type.

Parameters:

Name	Type	Description	Default
traces_list	list[dict[str, Any]]	The list of traces.	required

Returns:

Type	Description
tuple[float, str | None]	tuple[float, str | None]: The aggregated score and the first engine_type seen (if any).

check_langfuse_host_accessibility(langfuse_host)

Check if the Langfuse host is accessible.

Parameters:

Name	Type	Description	Default
langfuse_host	str | None	The Langfuse host.	required

Returns:

Name	Type	Description
bool	bool	True if the Langfuse host is accessible, False otherwise.

convert_to_langfuse_dataset(dataset, expected_output_key=DefaultValues.EXPECTED_OUTPUT_KEY, mapping=None)

Convert the dataset to Langfuse items.

Parameters:

Name	Type	Description	Default
dataset	LLMTestCase | dict[str, Any] | list[dict[str, Any] | LLMTestCase]	The dataset to convert.	required
expected_output_key	str | None	The key for expected output. Defaults to "expected_output".	EXPECTED_OUTPUT_KEY
mapping	dict[str, Any] | None	Optional mapping for field keys. Format: { "input": {"old_key": "new_key", "old_key_2": "new_key_2"}, "expected_output": "field_name_containing_expected_output" | {"old_key": "new_key"}, "metadata": {"old_key": "new_key", "old_key_2": "new_key_2"} }	None

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: The converted dataset.

convert_to_standard_dataset(dataset, expected_output_key=DefaultValues.EXPECTED_OUTPUT_KEY, mapping=None, convert_fixed_cols=False, fixed_cols_mapping=None)

Convert the dataset from Langfuse format back to standard format using reverse mapping.

Parameters:

Name	Type	Description	Default
dataset	dict[str, Any] | list[dict[str, Any]]	The dataset in Langfuse format.	required
expected_output_key	str | None	The key for expected output. Defaults to "expected_output".	EXPECTED_OUTPUT_KEY
mapping	dict[str, Any] | None	Optional mapping for field keys (same format as convert_to_langfuse_dataset).	None
convert_fixed_cols	bool	If True, convert the fixed columns mapping. Defaults to False.	False
fixed_cols_mapping	list[str] | None	Optional list of keys to skip. Defaults to FIXED_COLS_MAPPING.	None
Mapping format		{ "input": {"old_key": "new_key", "old_key_2": "new_key_2"}, "expected_output": "field_name_containing_expected_output" | {"old_key": "new_key"}, "metadata": {"old_key": "new_key", "old_key_2": "new_key_2"} }	required

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: The dataset in standard format.

create_langfuse_dataset(langfuse_client, dataset_name, dataset_description='', metadata=None)

Create a dataset in Langfuse.

Parameters:

Name	Type	Description	Default
langfuse_client	Langfuse	The Langfuse client.	required
dataset_name	str	The name of the dataset in Langfuse.	required
dataset_description	str	The description of the dataset.	''
metadata	dict[str, Any] | None	The metadata of the dataset.	None

Returns:

Name	Type	Description
Dataset	Dataset	The created dataset.

Raises:

Type	Description
ValueError	If dataset_name is empty or None.
ConnectionError	If there's a network connection issue.
RuntimeError	For other Langfuse API errors.

deanonymize_context(context, anonymizer_mappings)

De-anonymize the context.

Parameters:

Name	Type	Description	Default
context	str	The context to de-anonymize.	required
anonymizer_mappings	list[dict[str, str]]	The anonymizer mappings.	required

Returns:

Name	Type	Description
str	str	The de-anonymized context.

extract_key_value_from_observation_output(raw_string, key)

Extract the value of a key from a raw string.

1) Match a quoted string value (either '...' or "...") - \1 captures the quote used (single or double) - (?:\.|(?!\1).)*? matches any char, allowing escaped quotes like \" or \'

2) Match a list value: 'key': [ ... ]

Parameters:

Name	Type	Description	Default
raw_string	str	The raw string to extract the value from.	required
key	str	The key to extract the value from.	required

Returns:

Type	Description
str | None	str | None: The value of the key. None if the key is not found.

fetch_langfuse_traces_based_on_session_id(langfuse_client, session_id, other_search_params=None)

Fetch the Langfuse traces based on the session ID.

Parameters:

Name	Type	Description	Default
langfuse_client	Langfuse	The Langfuse client.	required
session_id	str	The session ID.	required
other_search_params	dict[str, Any] | None	The other parameters to pass to the search.	None

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: The list of traces as dictionaries with expanded scores.

find_or_update_dataset_item(langfuse_client, dataset_name, data, item_id=None, metadata=None, project_name='')

Find or update a dataset item in Langfuse.

This is a high-level function that ensures the dataset exists and the item exists. It combines dataset creation (if needed) and item creation/update/syncing.

Parameters:

Name	Type	Description	Default
langfuse_client	Langfuse	The Langfuse client.	required
dataset_name	str	The name of the dataset.	required
data	dict[str, Any]	The dataset data.	required
item_id	str | None	The ID of the dataset item.	None
metadata	dict[str, Any] | None	Additional metadata for dataset creation.	None
project_name	str	The project name for dataset creation. Defaults to "".	''

Returns:

Name	Type	Description
DatasetItemClient	DatasetItemClient	The dataset item.

Raises:

Type	Description
ValueError	If the dataset item cannot be found or created.

find_query_in_observations_output(langfuse_client, traces, observation_name, query)

Find the observation output of a trace that contains the specific query.

Parameters:

Name	Type	Description	Default
langfuse_client	Langfuse	The Langfuse client.	required
traces	Traces	The traces to search.	required
observation_name	str	The name of the observation to search.	required
query	str	The query to find.	required

Returns:

Type	Description
ObservationsView | None	ObservationsView | None: The observation from the trace.

get_dataset_run_url(langfuse_client, project_name, dataset_name, run_name)

Get the Langfuse URL for a specific dataset run.

Parameters:

Name	Type	Description	Default
langfuse_client	Langfuse	The Langfuse client.	required
project_name	str	The name of the Langfuse project.	required
dataset_name	str	The name of the Langfuse dataset.	required
run_name	str	The name of the Langfuse run.	required

Returns:

Type	Description
str | None	str | None: The Langfuse URL for the dataset run, or None if not found.

get_langfuse_dataset(langfuse_client, dataset_name)

Get a dataset from Langfuse if it exists.

Parameters:

Name	Type	Description	Default
langfuse_client	Langfuse	The Langfuse client.	required
dataset_name	str	The name of the dataset in Langfuse.	required

Returns:

Type	Description
DatasetClient | None	DatasetClient | None: The dataset if it exists, None otherwise.

get_langfuse_traces_expected_search_params()

Get all LangfuseTracesSearchParams as a list.

Returns:

Type	Description
list[str]	list[str]: list of parameter names.

get_observation_from_trace(langfuse_client, trace_id, observation_name)

Get the data from a trace.

Parameters:

Name	Type	Description	Default
langfuse_client	Langfuse	The Langfuse client.	required
trace_id	str	The ID of the trace.	required
observation_name	str	The name of the observation to get.	required

Returns:

Type	Description
ObservationsView | None	ObservationsView | None: The observation from the trace.

get_or_sync_langfuse_dataset_item(langfuse_client, dataset_name=None, dataset=None, item_id=None, data=None)

Get a dataset item from Langfuse and sync it if needed.

This function retrieves a dataset item from Langfuse. When searching by question ID (via the data parameter), it automatically updates the dataset item if discrepancies are detected between the existing item and the provided data. This ensures the dataset item is synchronized with the latest data.

The function requires at least the dataset name or dataset object, and either the item_id or the data dictionary which includes the question ID. If the dataset object is not provided, it will be fetched using the dataset name. The search is performed by direct ID if item_id is given, otherwise it searches by question ID contained in the data's input.

Parameters:

Name	Type	Description	Default
langfuse_client	Langfuse	The Langfuse client.	required
dataset_name	str | None	The name of the dataset in Langfuse.	None
dataset	DatasetClient | None	The dataset to get the item from.	None
item_id	str | None	The ID of the dataset item in Langfuse.	None
data	dict[str, Any] | None	The data of the dataset item in Langfuse. Must contain a question_id in the input when used for searching.	None

Returns:

Type	Description
DatasetItemClient | None	DatasetItemClient | None: The dataset item if it exists, None otherwise.

Raises:

Type	Description
ValueError	If neither dataset name nor dataset is provided, or neither item_id nor data is provided.

Note

⚠️ SIDE EFFECT: when searching by question ID (via data parameter), this function automatically updates the dataset item in Langfuse if the input, expected_output, or metadata differ from the existing item. This ensures data consistency but may modify the dataset item. When searching by item_id, no updates are performed.

Example

Read-only lookup by ID

item = get_or_sync_langfuse_dataset_item( ... client, "my_dataset", item_id="abc123" ... )

Lookup by question_id WITH automatic sync

item = get_or_sync_langfuse_dataset_item( ... client, "my_dataset", ... data={"input": {"question_id": "q1"}, ...} ... )

get_project_id(langfuse_client, project_name)

Get the project ID for a given project name.

Parameters:

Name	Type	Description	Default
langfuse_client	Langfuse	The Langfuse client.	required
project_name	str	The name of the Langfuse project.	required

Returns:

Type	Description
str | None	str | None: The project ID.

get_session_level_scores(langfuse_client, session_id, name=GeneralConstants.GENERATION, limit=DefaultValues.DEFAULT_LANGFUSE_SEARCH_TRACES_LIMIT, is_paginated=True)

Get the session level scores from Langfuse with pagination.

Parameters:

Name	Type	Description	Default
langfuse_client	Langfuse	The Langfuse client.	required
session_id	str	The session ID.	required
name	str	The name of the score. Defaults to GeneralConstants.GENERATION.	GENERATION
limit	int	The limit of the scores to get.	DEFAULT_LANGFUSE_SEARCH_TRACES_LIMIT
is_paginated	bool	Whether to use pagination to get the scores.	True

Returns:

Type	Description
str | None	str | None: The ID of the score if found, otherwise None.

get_session_url(langfuse_client, project_name, session_name)

Get the URL of a session.

Parameters:

Name	Type	Description	Default
langfuse_client	Langfuse	The Langfuse client.	required
project_name	str	The name of the project.	required
session_name	str	The session identifier (ID or human-friendly name).	required

Returns:

Type	Description
str | None	str | None: The URL of the session.

get_trace_url(langfuse_client, project_name, trace_id)

Get the URL of a trace.

Parameters:

Name	Type	Description	Default
langfuse_client	Langfuse	The Langfuse client.	required
project_name	str	The name of the project.	required
trace_id	str	The ID of the trace.	required

Returns:

Type	Description
str | None	str | None: The URL of the trace.

paginate_search_traces(langfuse_client, search_params)

Search for traces in Langfuse with pagination support.

Parameters:

Name	Type	Description	Default
langfuse_client	Langfuse	The Langfuse client.	required
search_params	dict[str, Any]	The search parameters to use.	required

Returns:

Name	Type	Description
Traces	Traces	The list of traces with all pages combined.

parse_anonymizer_mappings(raw_string)

Convert a string of [AnonymizerMapping(...), ...] into a list of dicts with the provided key/value pairs.

Parameters:

Name	Type	Description	Default
raw_string	str	The raw string to parse.	required

Example

"AnonymizerMapping(name='name', value='value')" "AnonymizerMapping(name='name', value='value')" ...

Returns:

Type	Description
list[dict[str, str]]	list[dict[str, str]]: The list of anonymizer mappings.

search_traces(langfuse_client, search_params)

Search for traces in Langfuse.

Parameters:

Name	Type	Description	Default
langfuse_client	Langfuse	The Langfuse client.	required
search_params	dict[str, Any]	The search params to use.	required

Returns:

Name	Type	Description
Traces	Traces	The traces.

upsert_score(langfuse_client, name, value, session_id=None, trace_id=None, score_id=None, comment=None)

Insert a score to session or trace level.

Parameters:

Name	Type	Description	Default
langfuse_client	Langfuse	The Langfuse client.	required
name	str	The name of the score.	required
value	float | str	The value of the score.	required
session_id	str | None	The session ID.	None
trace_id	str | None	The trace ID.	None
score_id	str | None	The score ID.	None
comment	str | None	The comment of the score.	None

Raises:

Type	Description
ValueError	If session_id, trace_id or score_id is not provided.

validate_search_params(search_params)

Validate the search params.

Parameters:

Name	Type	Description	Default
search_params	dict[str, Any]	The search params to validate.	required