From 121abc94d9fa5b48e7a3060574f7d511a15a3c9d Mon Sep 17 00:00:00 2001 From: Niklas Semmler <1880943+niklassemmler@users.noreply.github.com> Date: Wed, 7 Oct 2026 17:47:55 +0200 Subject: [PATCH 1/8] docs(api): document metadata is set / is not set filter operators (#1946) Regenerated langfuse/api from the upstream Fern definition: baseline langfuse/langfuse 619dbe3 (last sync, #1866) plus langfuse/langfuse#17357 and #18404. Split out of the bundled API-spec bot PR #1943. Co-authored-by: Claude Opus 5.5 (1M context) --- langfuse/api/__init__.py | 3 ++ langfuse/api/evaluation_commons/__init__.py | 3 ++ .../api/evaluation_commons/types/__init__.py | 5 +++ .../types/evaluation_rule_filter.py | 25 ++++++----- ...tion_rule_string_object_filter_operator.py | 42 +++++++++++++++++++ .../string_object_evaluation_rule_filter.py | 15 +++++-- langfuse/api/legacy/observations_v1/client.py | 4 +- .../api/legacy/observations_v1/raw_client.py | 4 +- langfuse/api/observations/client.py | 4 +- langfuse/api/observations/raw_client.py | 4 +- langfuse/api/scores/client.py | 4 +- langfuse/api/scores/raw_client.py | 4 +- langfuse/api/trace/client.py | 4 +- langfuse/api/trace/raw_client.py | 4 +- 14 files changed, 95 insertions(+), 30 deletions(-) create mode 100644 langfuse/api/evaluation_commons/types/evaluation_rule_string_object_filter_operator.py diff --git a/langfuse/api/__init__.py b/langfuse/api/__init__.py index ec4d25b20..275fdff7c 100644 --- a/langfuse/api/__init__.py +++ b/langfuse/api/__init__.py @@ -187,6 +187,7 @@ EvaluationRuleReadFilterBase, EvaluationRuleReadFilterWithKey, EvaluationRuleStringFilterOperator, + EvaluationRuleStringObjectFilterOperator, EvaluatorOutputDefinition, EvaluatorOutputDefinitionBase, EvaluatorOutputDefinition_Boolean, @@ -607,6 +608,7 @@ "EvaluationRuleReadFilterBase": ".evaluation_commons", "EvaluationRuleReadFilterWithKey": ".evaluation_commons", "EvaluationRuleStringFilterOperator": ".evaluation_commons", + "EvaluationRuleStringObjectFilterOperator": ".evaluation_commons", "EvaluationRulesPage": ".evaluation_rules", "EvaluationUnauthorizedError": ".evaluation_errors", "Evaluator": ".evaluators", @@ -1062,6 +1064,7 @@ def __dir__(): "EvaluationRuleReadFilterBase", "EvaluationRuleReadFilterWithKey", "EvaluationRuleStringFilterOperator", + "EvaluationRuleStringObjectFilterOperator", "EvaluationRulesPage", "EvaluationUnauthorizedError", "Evaluator", diff --git a/langfuse/api/evaluation_commons/__init__.py b/langfuse/api/evaluation_commons/__init__.py index 00912e7ce..e16b31a7c 100644 --- a/langfuse/api/evaluation_commons/__init__.py +++ b/langfuse/api/evaluation_commons/__init__.py @@ -32,6 +32,7 @@ EvaluationRuleReadFilterBase, EvaluationRuleReadFilterWithKey, EvaluationRuleStringFilterOperator, + EvaluationRuleStringObjectFilterOperator, EvaluatorOutputDefinition, EvaluatorOutputDefinitionBase, EvaluatorOutputDefinition_Boolean, @@ -85,6 +86,7 @@ "EvaluationRuleReadFilterBase": ".types", "EvaluationRuleReadFilterWithKey": ".types", "EvaluationRuleStringFilterOperator": ".types", + "EvaluationRuleStringObjectFilterOperator": ".types", "EvaluatorOutputDefinition": ".types", "EvaluatorOutputDefinitionBase": ".types", "EvaluatorOutputDefinition_Boolean": ".types", @@ -167,6 +169,7 @@ def __dir__(): "EvaluationRuleReadFilterBase", "EvaluationRuleReadFilterWithKey", "EvaluationRuleStringFilterOperator", + "EvaluationRuleStringObjectFilterOperator", "EvaluatorOutputDefinition", "EvaluatorOutputDefinitionBase", "EvaluatorOutputDefinition_Boolean", diff --git a/langfuse/api/evaluation_commons/types/__init__.py b/langfuse/api/evaluation_commons/types/__init__.py index d04011e88..7c5018f54 100644 --- a/langfuse/api/evaluation_commons/types/__init__.py +++ b/langfuse/api/evaluation_commons/types/__init__.py @@ -45,6 +45,9 @@ from .evaluation_rule_string_filter_operator import ( EvaluationRuleStringFilterOperator, ) + from .evaluation_rule_string_object_filter_operator import ( + EvaluationRuleStringObjectFilterOperator, + ) from .evaluator_output_definition import ( EvaluatorOutputDefinition, EvaluatorOutputDefinition_Boolean, @@ -101,6 +104,7 @@ "EvaluationRuleReadFilterBase": ".evaluation_rule_read_filter_base", "EvaluationRuleReadFilterWithKey": ".evaluation_rule_read_filter_with_key", "EvaluationRuleStringFilterOperator": ".evaluation_rule_string_filter_operator", + "EvaluationRuleStringObjectFilterOperator": ".evaluation_rule_string_object_filter_operator", "EvaluatorOutputDefinition": ".evaluator_output_definition", "EvaluatorOutputDefinitionBase": ".evaluator_output_definition_base", "EvaluatorOutputDefinition_Boolean": ".evaluator_output_definition", @@ -183,6 +187,7 @@ def __dir__(): "EvaluationRuleReadFilterBase", "EvaluationRuleReadFilterWithKey", "EvaluationRuleStringFilterOperator", + "EvaluationRuleStringObjectFilterOperator", "EvaluatorOutputDefinition", "EvaluatorOutputDefinitionBase", "EvaluatorOutputDefinition_Boolean", diff --git a/langfuse/api/evaluation_commons/types/evaluation_rule_filter.py b/langfuse/api/evaluation_commons/types/evaluation_rule_filter.py index f281dbb17..160026a20 100644 --- a/langfuse/api/evaluation_commons/types/evaluation_rule_filter.py +++ b/langfuse/api/evaluation_commons/types/evaluation_rule_filter.py @@ -16,6 +16,9 @@ from .evaluation_rule_number_filter_operator import EvaluationRuleNumberFilterOperator from .evaluation_rule_options_filter_operator import EvaluationRuleOptionsFilterOperator from .evaluation_rule_string_filter_operator import EvaluationRuleStringFilterOperator +from .evaluation_rule_string_object_filter_operator import ( + EvaluationRuleStringObjectFilterOperator, +) class EvaluationRuleFilter_Datetime(UniversalBaseModel): @@ -34,7 +37,7 @@ class EvaluationRuleFilter_Datetime(UniversalBaseModel): - `datetime`: `"="`, `">"`, `"<"`, `">="`, `"<="` - `stringOptions`: `any of`, `none of` - `arrayOptions`: `any of`, `none of`, `all of` - - `stringObject`: same operators as `string` + - `stringObject`: `"="`, `contains`, `does not contain`, `starts with`, `ends with`, `is set`, `is not set` (use `is set` / `is not set` for key presence; an empty value for `contains`, `starts with`, or `ends with` is treated as `is set`) - `boolean`: `"="`, `"<>"` - `null`: `is null`, `is not null` @@ -97,7 +100,7 @@ class EvaluationRuleFilter_String(UniversalBaseModel): - `datetime`: `"="`, `">"`, `"<"`, `">="`, `"<="` - `stringOptions`: `any of`, `none of` - `arrayOptions`: `any of`, `none of`, `all of` - - `stringObject`: same operators as `string` + - `stringObject`: `"="`, `contains`, `does not contain`, `starts with`, `ends with`, `is set`, `is not set` (use `is set` / `is not set` for key presence; an empty value for `contains`, `starts with`, or `ends with` is treated as `is set`) - `boolean`: `"="`, `"<>"` - `null`: `is null`, `is not null` @@ -160,7 +163,7 @@ class EvaluationRuleFilter_Number(UniversalBaseModel): - `datetime`: `"="`, `">"`, `"<"`, `">="`, `"<="` - `stringOptions`: `any of`, `none of` - `arrayOptions`: `any of`, `none of`, `all of` - - `stringObject`: same operators as `string` + - `stringObject`: `"="`, `contains`, `does not contain`, `starts with`, `ends with`, `is set`, `is not set` (use `is set` / `is not set` for key presence; an empty value for `contains`, `starts with`, or `ends with` is treated as `is set`) - `boolean`: `"="`, `"<>"` - `null`: `is null`, `is not null` @@ -223,7 +226,7 @@ class EvaluationRuleFilter_StringOptions(UniversalBaseModel): - `datetime`: `"="`, `">"`, `"<"`, `">="`, `"<="` - `stringOptions`: `any of`, `none of` - `arrayOptions`: `any of`, `none of`, `all of` - - `stringObject`: same operators as `string` + - `stringObject`: `"="`, `contains`, `does not contain`, `starts with`, `ends with`, `is set`, `is not set` (use `is set` / `is not set` for key presence; an empty value for `contains`, `starts with`, or `ends with` is treated as `is set`) - `boolean`: `"="`, `"<>"` - `null`: `is null`, `is not null` @@ -286,7 +289,7 @@ class EvaluationRuleFilter_CategoryOptions(UniversalBaseModel): - `datetime`: `"="`, `">"`, `"<"`, `">="`, `"<="` - `stringOptions`: `any of`, `none of` - `arrayOptions`: `any of`, `none of`, `all of` - - `stringObject`: same operators as `string` + - `stringObject`: `"="`, `contains`, `does not contain`, `starts with`, `ends with`, `is set`, `is not set` (use `is set` / `is not set` for key presence; an empty value for `contains`, `starts with`, or `ends with` is treated as `is set`) - `boolean`: `"="`, `"<>"` - `null`: `is null`, `is not null` @@ -350,7 +353,7 @@ class EvaluationRuleFilter_ArrayOptions(UniversalBaseModel): - `datetime`: `"="`, `">"`, `"<"`, `">="`, `"<="` - `stringOptions`: `any of`, `none of` - `arrayOptions`: `any of`, `none of`, `all of` - - `stringObject`: same operators as `string` + - `stringObject`: `"="`, `contains`, `does not contain`, `starts with`, `ends with`, `is set`, `is not set` (use `is set` / `is not set` for key presence; an empty value for `contains`, `starts with`, or `ends with` is treated as `is set`) - `boolean`: `"="`, `"<>"` - `null`: `is null`, `is not null` @@ -413,7 +416,7 @@ class EvaluationRuleFilter_StringObject(UniversalBaseModel): - `datetime`: `"="`, `">"`, `"<"`, `">="`, `"<="` - `stringOptions`: `any of`, `none of` - `arrayOptions`: `any of`, `none of`, `all of` - - `stringObject`: same operators as `string` + - `stringObject`: `"="`, `contains`, `does not contain`, `starts with`, `ends with`, `is set`, `is not set` (use `is set` / `is not set` for key presence; an empty value for `contains`, `starts with`, or `ends with` is treated as `is set`) - `boolean`: `"="`, `"<>"` - `null`: `is null`, `is not null` @@ -453,7 +456,7 @@ class EvaluationRuleFilter_StringObject(UniversalBaseModel): type: typing.Literal["stringObject"] = "stringObject" column: str key: str - operator: EvaluationRuleStringFilterOperator + operator: EvaluationRuleStringObjectFilterOperator value: str model_config: typing.ClassVar[pydantic.ConfigDict] = pydantic.ConfigDict( @@ -477,7 +480,7 @@ class EvaluationRuleFilter_NumberObject(UniversalBaseModel): - `datetime`: `"="`, `">"`, `"<"`, `">="`, `"<="` - `stringOptions`: `any of`, `none of` - `arrayOptions`: `any of`, `none of`, `all of` - - `stringObject`: same operators as `string` + - `stringObject`: `"="`, `contains`, `does not contain`, `starts with`, `ends with`, `is set`, `is not set` (use `is set` / `is not set` for key presence; an empty value for `contains`, `starts with`, or `ends with` is treated as `is set`) - `boolean`: `"="`, `"<>"` - `null`: `is null`, `is not null` @@ -541,7 +544,7 @@ class EvaluationRuleFilter_Boolean(UniversalBaseModel): - `datetime`: `"="`, `">"`, `"<"`, `">="`, `"<="` - `stringOptions`: `any of`, `none of` - `arrayOptions`: `any of`, `none of`, `all of` - - `stringObject`: same operators as `string` + - `stringObject`: `"="`, `contains`, `does not contain`, `starts with`, `ends with`, `is set`, `is not set` (use `is set` / `is not set` for key presence; an empty value for `contains`, `starts with`, or `ends with` is treated as `is set`) - `boolean`: `"="`, `"<>"` - `null`: `is null`, `is not null` @@ -604,7 +607,7 @@ class EvaluationRuleFilter_Null(UniversalBaseModel): - `datetime`: `"="`, `">"`, `"<"`, `">="`, `"<="` - `stringOptions`: `any of`, `none of` - `arrayOptions`: `any of`, `none of`, `all of` - - `stringObject`: same operators as `string` + - `stringObject`: `"="`, `contains`, `does not contain`, `starts with`, `ends with`, `is set`, `is not set` (use `is set` / `is not set` for key presence; an empty value for `contains`, `starts with`, or `ends with` is treated as `is set`) - `boolean`: `"="`, `"<>"` - `null`: `is null`, `is not null` diff --git a/langfuse/api/evaluation_commons/types/evaluation_rule_string_object_filter_operator.py b/langfuse/api/evaluation_commons/types/evaluation_rule_string_object_filter_operator.py new file mode 100644 index 000000000..edf675442 --- /dev/null +++ b/langfuse/api/evaluation_commons/types/evaluation_rule_string_object_filter_operator.py @@ -0,0 +1,42 @@ +# This file was auto-generated by Fern from our API Definition. + +import typing + +from ...core import enum + +T_Result = typing.TypeVar("T_Result") + + +class EvaluationRuleStringObjectFilterOperator(enum.StrEnum): + EQUALS = "=" + CONTAINS = "contains" + DOES_NOT_CONTAIN = "does not contain" + STARTS_WITH = "starts with" + ENDS_WITH = "ends with" + IS_SET = "is set" + IS_NOT_SET = "is not set" + + def visit( + self, + equals: typing.Callable[[], T_Result], + contains: typing.Callable[[], T_Result], + does_not_contain: typing.Callable[[], T_Result], + starts_with: typing.Callable[[], T_Result], + ends_with: typing.Callable[[], T_Result], + is_set: typing.Callable[[], T_Result], + is_not_set: typing.Callable[[], T_Result], + ) -> T_Result: + if self is EvaluationRuleStringObjectFilterOperator.EQUALS: + return equals() + if self is EvaluationRuleStringObjectFilterOperator.CONTAINS: + return contains() + if self is EvaluationRuleStringObjectFilterOperator.DOES_NOT_CONTAIN: + return does_not_contain() + if self is EvaluationRuleStringObjectFilterOperator.STARTS_WITH: + return starts_with() + if self is EvaluationRuleStringObjectFilterOperator.ENDS_WITH: + return ends_with() + if self is EvaluationRuleStringObjectFilterOperator.IS_SET: + return is_set() + if self is EvaluationRuleStringObjectFilterOperator.IS_NOT_SET: + return is_not_set() diff --git a/langfuse/api/evaluation_commons/types/string_object_evaluation_rule_filter.py b/langfuse/api/evaluation_commons/types/string_object_evaluation_rule_filter.py index d291153f0..c0aed0846 100644 --- a/langfuse/api/evaluation_commons/types/string_object_evaluation_rule_filter.py +++ b/langfuse/api/evaluation_commons/types/string_object_evaluation_rule_filter.py @@ -4,7 +4,9 @@ import pydantic from ...core.pydantic_utilities import UniversalBaseModel -from .evaluation_rule_string_filter_operator import EvaluationRuleStringFilterOperator +from .evaluation_rule_string_object_filter_operator import ( + EvaluationRuleStringObjectFilterOperator, +) class StringObjectEvaluationRuleFilter(UniversalBaseModel): @@ -18,8 +20,15 @@ class StringObjectEvaluationRuleFilter(UniversalBaseModel): Top-level key inside the object-valued column to filter on. """ - operator: EvaluationRuleStringFilterOperator - value: str + operator: EvaluationRuleStringObjectFilterOperator = pydantic.Field() + """ + Use `is set` / `is not set` to filter on key presence. + """ + + value: str = pydantic.Field() + """ + Value to compare against. Ignored for `is set` / `is not set`; send `""`. + """ model_config: typing.ClassVar[pydantic.ConfigDict] = pydantic.ConfigDict( extra="allow", frozen=True diff --git a/langfuse/api/legacy/observations_v1/client.py b/langfuse/api/legacy/observations_v1/client.py index 871191ee5..46ee7772a 100644 --- a/langfuse/api/legacy/observations_v1/client.py +++ b/langfuse/api/legacy/observations_v1/client.py @@ -150,7 +150,7 @@ def get_many( // - categoryOptions: "any of", "none of" // - arrayOptions: "any of", "none of", "all of" // - number: "=", ">", "<", ">=", "<=" - // - stringObject: "=", "contains", "does not contain", "starts with", "ends with" + // - stringObject: "=", "contains", "does not contain", "starts with", "ends with", "is set", "is not set" // - numberObject: "=", ">", "<", ">=", "<=" // - boolean: "=", "<>" // - null: "is null", "is not null" @@ -414,7 +414,7 @@ async def get_many( // - categoryOptions: "any of", "none of" // - arrayOptions: "any of", "none of", "all of" // - number: "=", ">", "<", ">=", "<=" - // - stringObject: "=", "contains", "does not contain", "starts with", "ends with" + // - stringObject: "=", "contains", "does not contain", "starts with", "ends with", "is set", "is not set" // - numberObject: "=", ">", "<", ">=", "<=" // - boolean: "=", "<>" // - null: "is null", "is not null" diff --git a/langfuse/api/legacy/observations_v1/raw_client.py b/langfuse/api/legacy/observations_v1/raw_client.py index 73343c130..ee1792052 100644 --- a/langfuse/api/legacy/observations_v1/raw_client.py +++ b/langfuse/api/legacy/observations_v1/raw_client.py @@ -211,7 +211,7 @@ def get_many( // - categoryOptions: "any of", "none of" // - arrayOptions: "any of", "none of", "all of" // - number: "=", ">", "<", ">=", "<=" - // - stringObject: "=", "contains", "does not contain", "starts with", "ends with" + // - stringObject: "=", "contains", "does not contain", "starts with", "ends with", "is set", "is not set" // - numberObject: "=", ">", "<", ">=", "<=" // - boolean: "=", "<>" // - null: "is null", "is not null" @@ -588,7 +588,7 @@ async def get_many( // - categoryOptions: "any of", "none of" // - arrayOptions: "any of", "none of", "all of" // - number: "=", ">", "<", ">=", "<=" - // - stringObject: "=", "contains", "does not contain", "starts with", "ends with" + // - stringObject: "=", "contains", "does not contain", "starts with", "ends with", "is set", "is not set" // - numberObject: "=", ">", "<", ">=", "<=" // - boolean: "=", "<>" // - null: "is null", "is not null" diff --git a/langfuse/api/observations/client.py b/langfuse/api/observations/client.py index 0bab55257..a44bd0707 100644 --- a/langfuse/api/observations/client.py +++ b/langfuse/api/observations/client.py @@ -153,7 +153,7 @@ def get_many( // - categoryOptions: "any of", "none of" // - arrayOptions: "any of", "none of", "all of" // - number: "=", ">", "<", ">=", "<=" - // - stringObject: "=", "contains", "does not contain", "starts with", "ends with", "matches" + // - stringObject: "=", "contains", "does not contain", "starts with", "ends with", "matches", "is set", "is not set" // - numberObject: "=", ">", "<", ">=", "<=" // - boolean: "=", "<>" // - null: "is null", "is not null" @@ -437,7 +437,7 @@ async def get_many( // - categoryOptions: "any of", "none of" // - arrayOptions: "any of", "none of", "all of" // - number: "=", ">", "<", ">=", "<=" - // - stringObject: "=", "contains", "does not contain", "starts with", "ends with", "matches" + // - stringObject: "=", "contains", "does not contain", "starts with", "ends with", "matches", "is set", "is not set" // - numberObject: "=", ">", "<", ">=", "<=" // - boolean: "=", "<>" // - null: "is null", "is not null" diff --git a/langfuse/api/observations/raw_client.py b/langfuse/api/observations/raw_client.py index a4d6b82e7..24a78cc02 100644 --- a/langfuse/api/observations/raw_client.py +++ b/langfuse/api/observations/raw_client.py @@ -151,7 +151,7 @@ def get_many( // - categoryOptions: "any of", "none of" // - arrayOptions: "any of", "none of", "all of" // - number: "=", ">", "<", ">=", "<=" - // - stringObject: "=", "contains", "does not contain", "starts with", "ends with", "matches" + // - stringObject: "=", "contains", "does not contain", "starts with", "ends with", "matches", "is set", "is not set" // - numberObject: "=", ">", "<", ">=", "<=" // - boolean: "=", "<>" // - null: "is null", "is not null" @@ -494,7 +494,7 @@ async def get_many( // - categoryOptions: "any of", "none of" // - arrayOptions: "any of", "none of", "all of" // - number: "=", ">", "<", ">=", "<=" - // - stringObject: "=", "contains", "does not contain", "starts with", "ends with", "matches" + // - stringObject: "=", "contains", "does not contain", "starts with", "ends with", "matches", "is set", "is not set" // - numberObject: "=", ">", "<", ">=", "<=" // - boolean: "=", "<>" // - null: "is null", "is not null" diff --git a/langfuse/api/scores/client.py b/langfuse/api/scores/client.py index d035dc0bd..95f94389a 100644 --- a/langfuse/api/scores/client.py +++ b/langfuse/api/scores/client.py @@ -231,7 +231,7 @@ def get_many( Comma-separated list of field groups to include in the response. Available field groups: 'score' (core score fields), 'trace' (trace properties: userId, tags, environment, sessionId). If not specified, both 'score' and 'trace' are returned by default. Example: 'score' to exclude trace data, 'score,trace' to include both. Note: When filtering by trace properties (using userId or traceTags parameters), the 'trace' field group must be included, otherwise a 400 error will be returned. filter : typing.Optional[str] - A JSON stringified array of filter objects. Each object requires type, column, operator, and value. Supports filtering by score metadata using the stringObject type. Example: [{"type":"stringObject","column":"metadata","key":"user_id","operator":"=","value":"abc123"}]. Supported types: stringObject (metadata key-value filtering), string, number, datetime, stringOptions, arrayOptions. Supported operators for stringObject: =, contains, does not contain, starts with, ends with. + A JSON stringified array of filter objects. Each object requires type, column, operator, and value. Supports filtering by score metadata using the stringObject type. Example: [{"type":"stringObject","column":"metadata","key":"user_id","operator":"=","value":"abc123"}]. Supported types: stringObject (metadata key-value filtering), string, number, datetime, stringOptions, arrayOptions. Supported operators for stringObject: =, contains, does not contain, starts with, ends with, is set, is not set. Use is set / is not set to filter on metadata key presence. An empty value for contains, starts with, or ends with is treated as is set (key presence). request_options : typing.Optional[RequestOptions] Request-specific configuration. @@ -544,7 +544,7 @@ async def get_many( Comma-separated list of field groups to include in the response. Available field groups: 'score' (core score fields), 'trace' (trace properties: userId, tags, environment, sessionId). If not specified, both 'score' and 'trace' are returned by default. Example: 'score' to exclude trace data, 'score,trace' to include both. Note: When filtering by trace properties (using userId or traceTags parameters), the 'trace' field group must be included, otherwise a 400 error will be returned. filter : typing.Optional[str] - A JSON stringified array of filter objects. Each object requires type, column, operator, and value. Supports filtering by score metadata using the stringObject type. Example: [{"type":"stringObject","column":"metadata","key":"user_id","operator":"=","value":"abc123"}]. Supported types: stringObject (metadata key-value filtering), string, number, datetime, stringOptions, arrayOptions. Supported operators for stringObject: =, contains, does not contain, starts with, ends with. + A JSON stringified array of filter objects. Each object requires type, column, operator, and value. Supports filtering by score metadata using the stringObject type. Example: [{"type":"stringObject","column":"metadata","key":"user_id","operator":"=","value":"abc123"}]. Supported types: stringObject (metadata key-value filtering), string, number, datetime, stringOptions, arrayOptions. Supported operators for stringObject: =, contains, does not contain, starts with, ends with, is set, is not set. Use is set / is not set to filter on metadata key presence. An empty value for contains, starts with, or ends with is treated as is set (key presence). request_options : typing.Optional[RequestOptions] Request-specific configuration. diff --git a/langfuse/api/scores/raw_client.py b/langfuse/api/scores/raw_client.py index b87521d7a..e4295f233 100644 --- a/langfuse/api/scores/raw_client.py +++ b/langfuse/api/scores/raw_client.py @@ -297,7 +297,7 @@ def get_many( Comma-separated list of field groups to include in the response. Available field groups: 'score' (core score fields), 'trace' (trace properties: userId, tags, environment, sessionId). If not specified, both 'score' and 'trace' are returned by default. Example: 'score' to exclude trace data, 'score,trace' to include both. Note: When filtering by trace properties (using userId or traceTags parameters), the 'trace' field group must be included, otherwise a 400 error will be returned. filter : typing.Optional[str] - A JSON stringified array of filter objects. Each object requires type, column, operator, and value. Supports filtering by score metadata using the stringObject type. Example: [{"type":"stringObject","column":"metadata","key":"user_id","operator":"=","value":"abc123"}]. Supported types: stringObject (metadata key-value filtering), string, number, datetime, stringOptions, arrayOptions. Supported operators for stringObject: =, contains, does not contain, starts with, ends with. + A JSON stringified array of filter objects. Each object requires type, column, operator, and value. Supports filtering by score metadata using the stringObject type. Example: [{"type":"stringObject","column":"metadata","key":"user_id","operator":"=","value":"abc123"}]. Supported types: stringObject (metadata key-value filtering), string, number, datetime, stringOptions, arrayOptions. Supported operators for stringObject: =, contains, does not contain, starts with, ends with, is set, is not set. Use is set / is not set to filter on metadata key presence. An empty value for contains, starts with, or ends with is treated as is set (key presence). request_options : typing.Optional[RequestOptions] Request-specific configuration. @@ -789,7 +789,7 @@ async def get_many( Comma-separated list of field groups to include in the response. Available field groups: 'score' (core score fields), 'trace' (trace properties: userId, tags, environment, sessionId). If not specified, both 'score' and 'trace' are returned by default. Example: 'score' to exclude trace data, 'score,trace' to include both. Note: When filtering by trace properties (using userId or traceTags parameters), the 'trace' field group must be included, otherwise a 400 error will be returned. filter : typing.Optional[str] - A JSON stringified array of filter objects. Each object requires type, column, operator, and value. Supports filtering by score metadata using the stringObject type. Example: [{"type":"stringObject","column":"metadata","key":"user_id","operator":"=","value":"abc123"}]. Supported types: stringObject (metadata key-value filtering), string, number, datetime, stringOptions, arrayOptions. Supported operators for stringObject: =, contains, does not contain, starts with, ends with. + A JSON stringified array of filter objects. Each object requires type, column, operator, and value. Supports filtering by score metadata using the stringObject type. Example: [{"type":"stringObject","column":"metadata","key":"user_id","operator":"=","value":"abc123"}]. Supported types: stringObject (metadata key-value filtering), string, number, datetime, stringOptions, arrayOptions. Supported operators for stringObject: =, contains, does not contain, starts with, ends with, is set, is not set. Use is set / is not set to filter on metadata key presence. An empty value for contains, starts with, or ends with is treated as is set (key presence). request_options : typing.Optional[RequestOptions] Request-specific configuration. diff --git a/langfuse/api/trace/client.py b/langfuse/api/trace/client.py index 2cf76a4a0..5674177c1 100644 --- a/langfuse/api/trace/client.py +++ b/langfuse/api/trace/client.py @@ -198,7 +198,7 @@ def list( // - categoryOptions: "any of", "none of" // - arrayOptions: "any of", "none of", "all of" // - number: "=", ">", "<", ">=", "<=" - // - stringObject: "=", "contains", "does not contain", "starts with", "ends with" + // - stringObject: "=", "contains", "does not contain", "starts with", "ends with", "is set", "is not set" // - numberObject: "=", ">", "<", ">=", "<=" // - booleanObject: "=", "<>" // - boolean: "=", "<>" @@ -574,7 +574,7 @@ async def list( // - categoryOptions: "any of", "none of" // - arrayOptions: "any of", "none of", "all of" // - number: "=", ">", "<", ">=", "<=" - // - stringObject: "=", "contains", "does not contain", "starts with", "ends with" + // - stringObject: "=", "contains", "does not contain", "starts with", "ends with", "is set", "is not set" // - numberObject: "=", ">", "<", ">=", "<=" // - booleanObject: "=", "<>" // - boolean: "=", "<>" diff --git a/langfuse/api/trace/raw_client.py b/langfuse/api/trace/raw_client.py index 7cb22a519..0797b9919 100644 --- a/langfuse/api/trace/raw_client.py +++ b/langfuse/api/trace/raw_client.py @@ -326,7 +326,7 @@ def list( // - categoryOptions: "any of", "none of" // - arrayOptions: "any of", "none of", "all of" // - number: "=", ">", "<", ">=", "<=" - // - stringObject: "=", "contains", "does not contain", "starts with", "ends with" + // - stringObject: "=", "contains", "does not contain", "starts with", "ends with", "is set", "is not set" // - numberObject: "=", ">", "<", ">=", "<=" // - booleanObject: "=", "<>" // - boolean: "=", "<>" @@ -938,7 +938,7 @@ async def list( // - categoryOptions: "any of", "none of" // - arrayOptions: "any of", "none of", "all of" // - number: "=", ">", "<", ">=", "<=" - // - stringObject: "=", "contains", "does not contain", "starts with", "ends with" + // - stringObject: "=", "contains", "does not contain", "starts with", "ends with", "is set", "is not set" // - numberObject: "=", ">", "<", ">=", "<=" // - booleanObject: "=", "<>" // - boolean: "=", "<>" From 195face2023e2525c3966f971f137657694a748f Mon Sep 17 00:00:00 2001 From: Niklas Semmler <1880943+niklassemmler@users.noreply.github.com> Date: Wed, 7 Oct 2026 19:00:52 +0200 Subject: [PATCH 2/8] fix(api): type Dataset JSON-schema response fields as objects (#1949) Regenerated langfuse/api from the upstream Fern definition: baseline langfuse/langfuse 619dbe3 (last sync, #1866) plus langfuse/langfuse#17892 and the datasets.yml/opentelemetry.yml part of #18406. Split out of the bundled API-spec bot PR #1943. Co-authored-by: Claude Opus 5.5 (1M context) --- langfuse/api/commons/types/dataset.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/langfuse/api/commons/types/dataset.py b/langfuse/api/commons/types/dataset.py index d312b291a..b7e1d6bfe 100644 --- a/langfuse/api/commons/types/dataset.py +++ b/langfuse/api/commons/types/dataset.py @@ -23,14 +23,16 @@ class Dataset(UniversalBaseModel): """ input_schema: typing_extensions.Annotated[ - typing.Optional[typing.Any], FieldMetadata(alias="inputSchema") + typing.Optional[typing.Dict[str, typing.Any]], + FieldMetadata(alias="inputSchema"), ] = pydantic.Field(default=None) """ JSON Schema for validating dataset item inputs """ expected_output_schema: typing_extensions.Annotated[ - typing.Optional[typing.Any], FieldMetadata(alias="expectedOutputSchema") + typing.Optional[typing.Dict[str, typing.Any]], + FieldMetadata(alias="expectedOutputSchema"), ] = pydantic.Field(default=None) """ JSON Schema for validating dataset item expected outputs From a01aa242ff1fc3880d6d292150c63efb723350b8 Mon Sep 17 00:00:00 2001 From: Niklas Semmler <1880943+niklassemmler@users.noreply.github.com> Date: Wed, 7 Oct 2026 19:05:14 +0200 Subject: [PATCH 3/8] refactor(api): share cursor pagination meta via CursorMetaResponse (#1948) Regenerated langfuse/api from the upstream Fern definition: baseline langfuse/langfuse 619dbe3 (last sync, #1866) plus langfuse/langfuse#17888. Split out of the bundled API-spec bot PR #1943. Co-authored-by: Claude Opus 5.5 (1M context) --- langfuse/api/evaluators/types/cursor_meta.py | 7 +++---- .../types/experiments_response_meta.py | 7 +++---- .../observations/types/observations_v2meta.py | 11 +++------- .../api/scores_v3/types/get_scores_v3meta.py | 8 ++----- langfuse/api/utils/__init__.py | 5 +++-- langfuse/api/utils/pagination/__init__.py | 9 +++++--- .../api/utils/pagination/types/__init__.py | 8 +++++-- .../pagination/types/cursor_meta_response.py | 21 +++++++++++++++++++ 8 files changed, 47 insertions(+), 29 deletions(-) create mode 100644 langfuse/api/utils/pagination/types/cursor_meta_response.py diff --git a/langfuse/api/evaluators/types/cursor_meta.py b/langfuse/api/evaluators/types/cursor_meta.py index 3ee6e1a0a..7e4c64a67 100644 --- a/langfuse/api/evaluators/types/cursor_meta.py +++ b/langfuse/api/evaluators/types/cursor_meta.py @@ -3,13 +3,12 @@ import typing import pydantic -from ...core.pydantic_utilities import UniversalBaseModel +from ...utils.pagination.types.cursor_meta_response import CursorMetaResponse -class CursorMeta(UniversalBaseModel): - cursor: typing.Optional[str] = pydantic.Field(default=None) +class CursorMeta(CursorMetaResponse): """ - Opaque cursor for the next page. Omitted when there is no next page. + Cursor pagination metadata. """ model_config: typing.ClassVar[pydantic.ConfigDict] = pydantic.ConfigDict( diff --git a/langfuse/api/experiments/types/experiments_response_meta.py b/langfuse/api/experiments/types/experiments_response_meta.py index 038bcc45d..61f2de4ea 100644 --- a/langfuse/api/experiments/types/experiments_response_meta.py +++ b/langfuse/api/experiments/types/experiments_response_meta.py @@ -3,13 +3,12 @@ import typing import pydantic -from ...core.pydantic_utilities import UniversalBaseModel +from ...utils.pagination.types.cursor_meta_response import CursorMetaResponse -class ExperimentsResponseMeta(UniversalBaseModel): - cursor: typing.Optional[str] = pydantic.Field(default=None) +class ExperimentsResponseMeta(CursorMetaResponse): """ - Versioned base64url cursor for retrieving the next page. Absent when there are no more results. + Metadata for cursor-based pagination. """ model_config: typing.ClassVar[pydantic.ConfigDict] = pydantic.ConfigDict( diff --git a/langfuse/api/observations/types/observations_v2meta.py b/langfuse/api/observations/types/observations_v2meta.py index 8f86a6512..46d9f42aa 100644 --- a/langfuse/api/observations/types/observations_v2meta.py +++ b/langfuse/api/observations/types/observations_v2meta.py @@ -3,17 +3,12 @@ import typing import pydantic -from ...core.pydantic_utilities import UniversalBaseModel +from ...utils.pagination.types.cursor_meta_response import CursorMetaResponse -class ObservationsV2Meta(UniversalBaseModel): +class ObservationsV2Meta(CursorMetaResponse): """ - Metadata for cursor-based pagination - """ - - cursor: typing.Optional[str] = pydantic.Field(default=None) - """ - Base64-encoded cursor to use for retrieving the next page. If not present, there are no more results. + Metadata for cursor-based pagination. """ model_config: typing.ClassVar[pydantic.ConfigDict] = pydantic.ConfigDict( diff --git a/langfuse/api/scores_v3/types/get_scores_v3meta.py b/langfuse/api/scores_v3/types/get_scores_v3meta.py index 7dfcfe0e1..832793364 100644 --- a/langfuse/api/scores_v3/types/get_scores_v3meta.py +++ b/langfuse/api/scores_v3/types/get_scores_v3meta.py @@ -3,15 +3,11 @@ import typing import pydantic -from ...core.pydantic_utilities import UniversalBaseModel +from ...utils.pagination.types.cursor_meta_response import CursorMetaResponse -class GetScoresV3Meta(UniversalBaseModel): +class GetScoresV3Meta(CursorMetaResponse): limit: int - cursor: typing.Optional[str] = pydantic.Field(default=None) - """ - URL-safe base64 (base64url) cursor for the next page. Absent when there are no more results. - """ model_config: typing.ClassVar[pydantic.ConfigDict] = pydantic.ConfigDict( extra="allow", frozen=True diff --git a/langfuse/api/utils/__init__.py b/langfuse/api/utils/__init__.py index b272f64b5..1690ee475 100644 --- a/langfuse/api/utils/__init__.py +++ b/langfuse/api/utils/__init__.py @@ -7,8 +7,9 @@ if typing.TYPE_CHECKING: from . import pagination - from .pagination import MetaResponse + from .pagination import CursorMetaResponse, MetaResponse _dynamic_imports: typing.Dict[str, str] = { + "CursorMetaResponse": ".pagination", "MetaResponse": ".pagination", "pagination": ".pagination", } @@ -41,4 +42,4 @@ def __dir__(): return sorted(lazy_attrs) -__all__ = ["MetaResponse", "pagination"] +__all__ = ["CursorMetaResponse", "MetaResponse", "pagination"] diff --git a/langfuse/api/utils/pagination/__init__.py b/langfuse/api/utils/pagination/__init__.py index 50821832d..59cdf0321 100644 --- a/langfuse/api/utils/pagination/__init__.py +++ b/langfuse/api/utils/pagination/__init__.py @@ -6,8 +6,11 @@ from importlib import import_module if typing.TYPE_CHECKING: - from .types import MetaResponse -_dynamic_imports: typing.Dict[str, str] = {"MetaResponse": ".types"} + from .types import CursorMetaResponse, MetaResponse +_dynamic_imports: typing.Dict[str, str] = { + "CursorMetaResponse": ".types", + "MetaResponse": ".types", +} def __getattr__(attr_name: str) -> typing.Any: @@ -37,4 +40,4 @@ def __dir__(): return sorted(lazy_attrs) -__all__ = ["MetaResponse"] +__all__ = ["CursorMetaResponse", "MetaResponse"] diff --git a/langfuse/api/utils/pagination/types/__init__.py b/langfuse/api/utils/pagination/types/__init__.py index 5c0d83028..85d13aee0 100644 --- a/langfuse/api/utils/pagination/types/__init__.py +++ b/langfuse/api/utils/pagination/types/__init__.py @@ -6,8 +6,12 @@ from importlib import import_module if typing.TYPE_CHECKING: + from .cursor_meta_response import CursorMetaResponse from .meta_response import MetaResponse -_dynamic_imports: typing.Dict[str, str] = {"MetaResponse": ".meta_response"} +_dynamic_imports: typing.Dict[str, str] = { + "CursorMetaResponse": ".cursor_meta_response", + "MetaResponse": ".meta_response", +} def __getattr__(attr_name: str) -> typing.Any: @@ -37,4 +41,4 @@ def __dir__(): return sorted(lazy_attrs) -__all__ = ["MetaResponse"] +__all__ = ["CursorMetaResponse", "MetaResponse"] diff --git a/langfuse/api/utils/pagination/types/cursor_meta_response.py b/langfuse/api/utils/pagination/types/cursor_meta_response.py new file mode 100644 index 000000000..eb30740ac --- /dev/null +++ b/langfuse/api/utils/pagination/types/cursor_meta_response.py @@ -0,0 +1,21 @@ +# This file was auto-generated by Fern from our API Definition. + +import typing + +import pydantic +from ....core.pydantic_utilities import UniversalBaseModel + + +class CursorMetaResponse(UniversalBaseModel): + """ + Metadata for cursor-based pagination. + """ + + cursor: typing.Optional[str] = pydantic.Field(default=None) + """ + Opaque cursor for the next page. Omitted when there is no next page. + """ + + model_config: typing.ClassVar[pydantic.ConfigDict] = pydantic.ConfigDict( + extra="allow", frozen=True + ) From a4cbc213f667cee8b24de437d5e0370747bc5679 Mon Sep 17 00:00:00 2001 From: Niklas Semmler <1880943+niklassemmler@users.noreply.github.com> Date: Wed, 7 Oct 2026 19:10:09 +0200 Subject: [PATCH 4/8] docs(api): document fromStartTime as required on experiments endpoints (#1947) * docs(api): document fromStartTime as required on experiments endpoints Regenerated langfuse/api from the upstream Fern definition: baseline langfuse/langfuse 619dbe3 (last sync, #1866) plus langfuse/langfuse#17603. Split out of the bundled API-spec bot PR #1943. Co-Authored-By: Claude Opus 5.5 (1M context) * docs(api): use language-neutral wording for experiments time window Regenerated from langfuse/langfuse branch docs/experiments-from-start-time-neutral-wording. The fromStartTime docstrings named `toStartTime`, but the Python parameter is `to_start_time`; they now say "an upper bound" instead. Co-Authored-By: Claude Opus 5.5 (1M context) --------- Co-authored-by: Claude Opus 5.5 (1M context) --- langfuse/api/experiments/client.py | 20 ++++++++++++++++---- langfuse/api/experiments/raw_client.py | 20 ++++++++++++++++---- 2 files changed, 32 insertions(+), 8 deletions(-) diff --git a/langfuse/api/experiments/client.py b/langfuse/api/experiments/client.py index a3d469217..66a90d95b 100644 --- a/langfuse/api/experiments/client.py +++ b/langfuse/api/experiments/client.py @@ -47,7 +47,10 @@ def list( Parameters ---------- from_start_time : dt.datetime - Retrieve only experiments on or after this datetime. + Required. Retrieve only experiments with events on or after this + datetime. A lower bound is required so the query stays fast on + large projects; pair it with an upper bound to select a specific + window. fields : typing.Optional[str] Comma-separated list of field groups to include. Available groups: @@ -145,7 +148,10 @@ def list_items( Parameters ---------- from_start_time : dt.datetime - Retrieve only experiment items started on or after this datetime. + Required. Retrieve only experiment items started on or after this + datetime. A lower bound is required so the query stays fast on + large projects; pair it with an upper bound to select a specific + window. fields : typing.Optional[str] Comma-separated list of field groups to include. Available groups: @@ -263,7 +269,10 @@ async def list( Parameters ---------- from_start_time : dt.datetime - Retrieve only experiments on or after this datetime. + Required. Retrieve only experiments with events on or after this + datetime. A lower bound is required so the query stays fast on + large projects; pair it with an upper bound to select a specific + window. fields : typing.Optional[str] Comma-separated list of field groups to include. Available groups: @@ -368,7 +377,10 @@ async def list_items( Parameters ---------- from_start_time : dt.datetime - Retrieve only experiment items started on or after this datetime. + Required. Retrieve only experiment items started on or after this + datetime. A lower bound is required so the query stays fast on + large projects; pair it with an upper bound to select a specific + window. fields : typing.Optional[str] Comma-separated list of field groups to include. Available groups: diff --git a/langfuse/api/experiments/raw_client.py b/langfuse/api/experiments/raw_client.py index 8736786df..189c826ba 100644 --- a/langfuse/api/experiments/raw_client.py +++ b/langfuse/api/experiments/raw_client.py @@ -45,7 +45,10 @@ def list( Parameters ---------- from_start_time : dt.datetime - Retrieve only experiments on or after this datetime. + Required. Retrieve only experiments with events on or after this + datetime. A lower bound is required so the query stays fast on + large projects; pair it with an upper bound to select a specific + window. fields : typing.Optional[str] Comma-separated list of field groups to include. Available groups: @@ -205,7 +208,10 @@ def list_items( Parameters ---------- from_start_time : dt.datetime - Retrieve only experiment items started on or after this datetime. + Required. Retrieve only experiment items started on or after this + datetime. A lower bound is required so the query stays fast on + large projects; pair it with an upper bound to select a specific + window. fields : typing.Optional[str] Comma-separated list of field groups to include. Available groups: @@ -374,7 +380,10 @@ async def list( Parameters ---------- from_start_time : dt.datetime - Retrieve only experiments on or after this datetime. + Required. Retrieve only experiments with events on or after this + datetime. A lower bound is required so the query stays fast on + large projects; pair it with an upper bound to select a specific + window. fields : typing.Optional[str] Comma-separated list of field groups to include. Available groups: @@ -534,7 +543,10 @@ async def list_items( Parameters ---------- from_start_time : dt.datetime - Retrieve only experiment items started on or after this datetime. + Required. Retrieve only experiment items started on or after this + datetime. A lower bound is required so the query stays fast on + large projects; pair it with an upper bound to select a specific + window. fields : typing.Optional[str] Comma-separated list of field groups to include. Available groups: From bf11ec121479145923ff27b32ec2ad1d06fd2d51 Mon Sep 17 00:00:00 2001 From: Niklas Semmler <1880943+niklassemmler@users.noreply.github.com> Date: Wed, 7 Oct 2026 19:14:50 +0200 Subject: [PATCH 5/8] fix(api): type read-side metadata as optional objects (#1950) * fix(api): type read-side metadata as optional objects Regenerated langfuse/api from the upstream Fern definition: baseline langfuse/langfuse 619dbe3 (last sync, #1866) plus langfuse/langfuse#18122 and the commons.yml part of #18406. Split out of the bundled API-spec bot PR #1943. Co-Authored-By: Claude Opus 5.5 (1M context) * test(api): pin read-side metadata parsing on generated models Cover omitted, null, empty and nested-JSON metadata plus rejection of non-object metadata for Trace, Observation, ObservationV2, BaseScore, BaseScoreV1 and every Score / ScoreV1 / GetScoresResponseData variant. Co-Authored-By: Claude Opus 5.5 (1M context) --------- Co-authored-by: Claude Opus 5.5 (1M context) --- langfuse/api/commons/types/base_score.py | 4 +- langfuse/api/commons/types/base_score_v1.py | 4 +- langfuse/api/commons/types/observation.py | 4 +- langfuse/api/commons/types/observation_v2.py | 4 +- langfuse/api/commons/types/score.py | 10 +- langfuse/api/commons/types/score_v1.py | 8 +- langfuse/api/commons/types/trace.py | 6 +- .../scores/types/get_scores_response_data.py | 10 +- tests/unit/test_api_metadata_parsing.py | 166 ++++++++++++++++++ 9 files changed, 196 insertions(+), 20 deletions(-) create mode 100644 tests/unit/test_api_metadata_parsing.py diff --git a/langfuse/api/commons/types/base_score.py b/langfuse/api/commons/types/base_score.py index 44e09033c..f46240dae 100644 --- a/langfuse/api/commons/types/base_score.py +++ b/langfuse/api/commons/types/base_score.py @@ -61,7 +61,9 @@ class BaseScore(UniversalBaseModel): Comment on the score """ - metadata: typing.Any = pydantic.Field() + metadata: typing.Optional[typing.Dict[str, typing.Any]] = pydantic.Field( + default=None + ) """ Metadata associated with the score """ diff --git a/langfuse/api/commons/types/base_score_v1.py b/langfuse/api/commons/types/base_score_v1.py index 881b10a3b..13845d997 100644 --- a/langfuse/api/commons/types/base_score_v1.py +++ b/langfuse/api/commons/types/base_score_v1.py @@ -41,7 +41,9 @@ class BaseScoreV1(UniversalBaseModel): Comment on the score """ - metadata: typing.Any = pydantic.Field() + metadata: typing.Optional[typing.Dict[str, typing.Any]] = pydantic.Field( + default=None + ) """ Metadata associated with the score """ diff --git a/langfuse/api/commons/types/observation.py b/langfuse/api/commons/types/observation.py index 91bd50643..76c019b0d 100644 --- a/langfuse/api/commons/types/observation.py +++ b/langfuse/api/commons/types/observation.py @@ -77,7 +77,9 @@ class Observation(UniversalBaseModel): The version of the observation """ - metadata: typing.Any = pydantic.Field() + metadata: typing.Optional[typing.Dict[str, typing.Any]] = pydantic.Field( + default=None + ) """ Additional metadata of the observation """ diff --git a/langfuse/api/commons/types/observation_v2.py b/langfuse/api/commons/types/observation_v2.py index f034d8ace..9c6d9e3e6 100644 --- a/langfuse/api/commons/types/observation_v2.py +++ b/langfuse/api/commons/types/observation_v2.py @@ -152,7 +152,9 @@ class ObservationV2(UniversalBaseModel): The output data of the observation """ - metadata: typing.Optional[typing.Any] = pydantic.Field(default=None) + metadata: typing.Optional[typing.Dict[str, typing.Any]] = pydantic.Field( + default=None + ) """ Additional metadata of the observation """ diff --git a/langfuse/api/commons/types/score.py b/langfuse/api/commons/types/score.py index 7b31ae39a..e0e98c9ae 100644 --- a/langfuse/api/commons/types/score.py +++ b/langfuse/api/commons/types/score.py @@ -54,7 +54,7 @@ class Score_Numeric(Base): typing.Optional[str], FieldMetadata(alias="authorUserId") ] = None comment: typing.Optional[str] = None - metadata: typing.Any + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None config_id: typing_extensions.Annotated[ typing.Optional[str], FieldMetadata(alias="configId") ] = None @@ -100,7 +100,7 @@ class Score_Categorical(Base): typing.Optional[str], FieldMetadata(alias="authorUserId") ] = None comment: typing.Optional[str] = None - metadata: typing.Any + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None config_id: typing_extensions.Annotated[ typing.Optional[str], FieldMetadata(alias="configId") ] = None @@ -146,7 +146,7 @@ class Score_Boolean(Base): typing.Optional[str], FieldMetadata(alias="authorUserId") ] = None comment: typing.Optional[str] = None - metadata: typing.Any + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None config_id: typing_extensions.Annotated[ typing.Optional[str], FieldMetadata(alias="configId") ] = None @@ -192,7 +192,7 @@ class Score_Correction(Base): typing.Optional[str], FieldMetadata(alias="authorUserId") ] = None comment: typing.Optional[str] = None - metadata: typing.Any + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None config_id: typing_extensions.Annotated[ typing.Optional[str], FieldMetadata(alias="configId") ] = None @@ -237,7 +237,7 @@ class Score_Text(Base): typing.Optional[str], FieldMetadata(alias="authorUserId") ] = None comment: typing.Optional[str] = None - metadata: typing.Any + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None config_id: typing_extensions.Annotated[ typing.Optional[str], FieldMetadata(alias="configId") ] = None diff --git a/langfuse/api/commons/types/score_v1.py b/langfuse/api/commons/types/score_v1.py index e17e61b89..32e0a6b4e 100644 --- a/langfuse/api/commons/types/score_v1.py +++ b/langfuse/api/commons/types/score_v1.py @@ -35,7 +35,7 @@ class ScoreV1_Numeric(UniversalBaseModel): typing.Optional[str], FieldMetadata(alias="authorUserId") ] = None comment: typing.Optional[str] = None - metadata: typing.Any + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None config_id: typing_extensions.Annotated[ typing.Optional[str], FieldMetadata(alias="configId") ] = None @@ -73,7 +73,7 @@ class ScoreV1_Categorical(UniversalBaseModel): typing.Optional[str], FieldMetadata(alias="authorUserId") ] = None comment: typing.Optional[str] = None - metadata: typing.Any + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None config_id: typing_extensions.Annotated[ typing.Optional[str], FieldMetadata(alias="configId") ] = None @@ -111,7 +111,7 @@ class ScoreV1_Boolean(UniversalBaseModel): typing.Optional[str], FieldMetadata(alias="authorUserId") ] = None comment: typing.Optional[str] = None - metadata: typing.Any + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None config_id: typing_extensions.Annotated[ typing.Optional[str], FieldMetadata(alias="configId") ] = None @@ -148,7 +148,7 @@ class ScoreV1_Text(UniversalBaseModel): typing.Optional[str], FieldMetadata(alias="authorUserId") ] = None comment: typing.Optional[str] = None - metadata: typing.Any + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None config_id: typing_extensions.Annotated[ typing.Optional[str], FieldMetadata(alias="configId") ] = None diff --git a/langfuse/api/commons/types/trace.py b/langfuse/api/commons/types/trace.py index 6eafa9fcd..1b489c03c 100644 --- a/langfuse/api/commons/types/trace.py +++ b/langfuse/api/commons/types/trace.py @@ -59,9 +59,11 @@ class Trace(UniversalBaseModel): The user identifier associated with the trace """ - metadata: typing.Optional[typing.Any] = pydantic.Field(default=None) + metadata: typing.Optional[typing.Dict[str, typing.Any]] = pydantic.Field( + default=None + ) """ - The metadata associated with the trace. Can be any JSON. + The metadata associated with the trace. Values can be any JSON; non-object metadata sent at ingestion is returned under the `metadata` key. """ tags: typing.List[str] = pydantic.Field() diff --git a/langfuse/api/scores/types/get_scores_response_data.py b/langfuse/api/scores/types/get_scores_response_data.py index f85cd471c..84b8abcc4 100644 --- a/langfuse/api/scores/types/get_scores_response_data.py +++ b/langfuse/api/scores/types/get_scores_response_data.py @@ -45,7 +45,7 @@ class GetScoresResponseData_Numeric(UniversalBaseModel): typing.Optional[str], FieldMetadata(alias="authorUserId") ] = None comment: typing.Optional[str] = None - metadata: typing.Any + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None config_id: typing_extensions.Annotated[ typing.Optional[str], FieldMetadata(alias="configId") ] = None @@ -92,7 +92,7 @@ class GetScoresResponseData_Categorical(UniversalBaseModel): typing.Optional[str], FieldMetadata(alias="authorUserId") ] = None comment: typing.Optional[str] = None - metadata: typing.Any + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None config_id: typing_extensions.Annotated[ typing.Optional[str], FieldMetadata(alias="configId") ] = None @@ -139,7 +139,7 @@ class GetScoresResponseData_Boolean(UniversalBaseModel): typing.Optional[str], FieldMetadata(alias="authorUserId") ] = None comment: typing.Optional[str] = None - metadata: typing.Any + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None config_id: typing_extensions.Annotated[ typing.Optional[str], FieldMetadata(alias="configId") ] = None @@ -186,7 +186,7 @@ class GetScoresResponseData_Correction(UniversalBaseModel): typing.Optional[str], FieldMetadata(alias="authorUserId") ] = None comment: typing.Optional[str] = None - metadata: typing.Any + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None config_id: typing_extensions.Annotated[ typing.Optional[str], FieldMetadata(alias="configId") ] = None @@ -232,7 +232,7 @@ class GetScoresResponseData_Text(UniversalBaseModel): typing.Optional[str], FieldMetadata(alias="authorUserId") ] = None comment: typing.Optional[str] = None - metadata: typing.Any + metadata: typing.Optional[typing.Dict[str, typing.Any]] = None config_id: typing_extensions.Annotated[ typing.Optional[str], FieldMetadata(alias="configId") ] = None diff --git a/tests/unit/test_api_metadata_parsing.py b/tests/unit/test_api_metadata_parsing.py new file mode 100644 index 000000000..6be15dfba --- /dev/null +++ b/tests/unit/test_api_metadata_parsing.py @@ -0,0 +1,166 @@ +"""Response parsing of read-side ``metadata`` on generated API models. + +The server returns read-side metadata as a JSON object, ``null``, or omits it. +These tests pin that contract so a future regeneration that makes the field +required or narrows its values is caught. +""" + +import typing + +import pydantic +import pytest + +from langfuse.api import ( + BaseScore, + BaseScoreV1, + GetScoresResponseData, + Observation, + ObservationV2, + Score, + ScoreV1, + Trace, +) +from langfuse.api.core import parse_obj_as + +TIMESTAMP = "2026-01-01T00:00:00.000Z" + +SCORE_BASE: typing.Dict[str, typing.Any] = { + "id": "score-1", + "name": "quality", + "source": "API", + "timestamp": TIMESTAMP, + "createdAt": TIMESTAMP, + "updatedAt": TIMESTAMP, + "environment": "default", +} +SCORE_V1_BASE = {**SCORE_BASE, "traceId": "trace-1"} + +SCORE_VARIANTS: typing.Dict[str, typing.Dict[str, typing.Any]] = { + "NUMERIC": {"value": 0.5}, + "CATEGORICAL": {"value": 1, "stringValue": "good"}, + "BOOLEAN": {"value": 1, "stringValue": "True"}, + "CORRECTION": {"value": 0, "stringValue": "corrected output"}, + "TEXT": {"stringValue": "free text"}, +} +SCORE_V1_DATA_TYPES = ["NUMERIC", "CATEGORICAL", "BOOLEAN", "TEXT"] + + +def _score_payload( + base: typing.Dict[str, typing.Any], data_type: str +) -> typing.Dict[str, typing.Any]: + return {**base, "dataType": data_type, **SCORE_VARIANTS[data_type]} + + +CASES: typing.List[typing.Any] = [ + pytest.param( + Trace, + { + "id": "trace-1", + "timestamp": TIMESTAMP, + "tags": [], + "public": False, + "environment": "default", + }, + id="Trace", + ), + pytest.param( + Observation, + { + "id": "obs-1", + "type": "SPAN", + "startTime": TIMESTAMP, + "modelParameters": {}, + "input": None, + "output": None, + "usage": {"input": 0, "output": 0, "total": 0}, + "level": "DEFAULT", + "usageDetails": {}, + "costDetails": {}, + "environment": "default", + }, + id="Observation", + ), + pytest.param( + ObservationV2, + { + "id": "obs-1", + "startTime": TIMESTAMP, + "projectId": "project-1", + "type": "SPAN", + }, + id="ObservationV2", + ), + pytest.param(BaseScore, SCORE_BASE, id="BaseScore"), + pytest.param(BaseScoreV1, SCORE_V1_BASE, id="BaseScoreV1"), + *[ + pytest.param(Score, _score_payload(SCORE_BASE, dt), id=f"Score-{dt}") + for dt in SCORE_VARIANTS + ], + *[ + pytest.param( + GetScoresResponseData, + _score_payload(SCORE_BASE, dt), + id=f"GetScoresResponseData-{dt}", + ) + for dt in SCORE_VARIANTS + ], + *[ + pytest.param(ScoreV1, _score_payload(SCORE_V1_BASE, dt), id=f"ScoreV1-{dt}") + for dt in SCORE_V1_DATA_TYPES + ], +] + +NESTED_METADATA = { + "str": "value", + "int": 1, + "float": 1.5, + "bool": True, + "null": None, + "list": [1, "two", {"three": 3}], + "nested": {"deeper": {"key": ["a", "b"]}}, +} + + +@pytest.mark.parametrize(("type_", "payload"), CASES) +def test_metadata_omitted_parses_as_none(type_, payload): + assert "metadata" not in payload + + parsed = parse_obj_as(type_, payload) + + assert parsed.metadata is None + + +@pytest.mark.parametrize(("type_", "payload"), CASES) +def test_metadata_null_parses_as_none(type_, payload): + parsed = parse_obj_as(type_, {**payload, "metadata": None}) + + assert parsed.metadata is None + + +@pytest.mark.parametrize(("type_", "payload"), CASES) +def test_metadata_empty_object_parses(type_, payload): + parsed = parse_obj_as(type_, {**payload, "metadata": {}}) + + assert parsed.metadata == {} + + +@pytest.mark.parametrize(("type_", "payload"), CASES) +def test_metadata_object_with_nested_json_values_parses(type_, payload): + parsed = parse_obj_as(type_, {**payload, "metadata": NESTED_METADATA}) + + assert parsed.metadata == NESTED_METADATA + + +@pytest.mark.parametrize(("type_", "payload"), CASES) +@pytest.mark.parametrize( + "metadata", + ["a string", 1, 1.5, True, ["a", "list"]], + ids=["str", "int", "float", "bool", "list"], +) +def test_metadata_non_object_is_rejected(type_, payload, metadata): + with pytest.raises(pydantic.ValidationError) as exc_info: + parse_obj_as(type_, {**payload, "metadata": metadata}) + + # The discriminated unions prefix the location with the variant tag. + assert exc_info.value.errors() + assert all("metadata" in error["loc"] for error in exc_info.value.errors()) From 47c4ee4fc2dae3911647e7f2f1c2bbb6531d4b3c Mon Sep 17 00:00:00 2001 From: po-et <42566883+po-et@users.noreply.github.com> Date: Mon, 12 Oct 2026 15:34:05 +0800 Subject: [PATCH 6/8] fix(experiments): run synchronous tasks outside the event loop --- langfuse/experiment.py | 39 +++++- tests/unit/test_experiment.py | 231 ++++++++++++++++++++++++++++++++++ 2 files changed, 269 insertions(+), 1 deletion(-) diff --git a/langfuse/experiment.py b/langfuse/experiment.py index aa5481829..848846e87 100644 --- a/langfuse/experiment.py +++ b/langfuse/experiment.py @@ -6,7 +6,9 @@ """ import asyncio +import inspect from datetime import datetime +from functools import partial from typing import ( TYPE_CHECKING, Any, @@ -607,6 +609,10 @@ class TaskFunction(Protocol): - Return any type of output (will be passed to evaluators) - Can be either synchronous or asynchronous - Should handle their own errors gracefully (exceptions will be logged) + + Synchronous task functions run in worker threads so blocking calls can run + concurrently. Asynchronous task functions, including their decorated wrappers, + run on the experiment's event loop. """ def __call__( @@ -1021,9 +1027,40 @@ async def _run_evaluator( return [] +def _is_async_task(task: Any) -> bool: + """Inspect async task metadata without replacing the callable we invoke.""" + seen: set[int] = set() + check_call = True + for _ in range(100): + if id(task) in seen: + return False + seen.add(id(task)) + if isinstance(task, partial): + task = task.func + continue + try: + unwrapped = inspect.unwrap(task, stop=inspect.iscoroutinefunction) + except ValueError: + return False + if inspect.iscoroutinefunction(unwrapped): + return True + if unwrapped is not task: + task = unwrapped + elif check_call: + task = getattr(task, "__call__", None) + check_call = False + else: + return False + return False + + async def _run_task(task: TaskFunction, item: ExperimentItem) -> Any: """Run a task function and handle sync/async.""" - result = task(item=item) + result: Any + if _is_async_task(task): + result = task(item=item) + else: + result = await asyncio.to_thread(task, item=item) # Handle async tasks if asyncio.iscoroutine(result): diff --git a/tests/unit/test_experiment.py b/tests/unit/test_experiment.py index 3eb55ad40..b6ee45f1e 100644 --- a/tests/unit/test_experiment.py +++ b/tests/unit/test_experiment.py @@ -1,18 +1,23 @@ """Tests for ``langfuse.experiment`` — ``RunnerContext`` and ``RegressionError``.""" +import asyncio import inspect import typing from datetime import datetime +from functools import partial, wraps +from threading import Barrier, get_ident from typing import get_type_hints from unittest.mock import MagicMock import pytest +from opentelemetry import context as otel_context from opentelemetry import trace as otel_trace_api from langfuse import Evaluation, RegressionError, RunnerContext from langfuse._client.attributes import LangfuseOtelSpanAttributes from langfuse._client.client import Langfuse from langfuse.batch_evaluation import CompositeEvaluatorFunction +from langfuse.experiment import _run_task def _noop_task(*, item, **kwargs): # pragma: no cover - never invoked via mock @@ -25,6 +30,66 @@ def _make_ctx(**kwargs) -> RunnerContext: return RunnerContext(client=client, **kwargs) +def _task_form(task, form): + if "callable" in form: + wrapped_task = task + + class CallableTask: + @wraps(wrapped_task) + def __call__(self, **kwargs): + return wrapped_task(**kwargs) + + task = CallableTask() + + if "partial" in form: + task = partial(task, prefix="partial") + + return task + + +@pytest.mark.asyncio +async def test_async_task_stays_on_experiment_event_loop(): + loop = asyncio.get_running_loop() + + async def task(*, item): + assert asyncio.get_running_loop() is loop + return item["input"] + + assert await _run_task(task, {"input": "async-result"}) == "async-result" + + +@pytest.mark.asyncio +async def test_sync_task_returning_coroutine_is_awaited(): + loop = asyncio.get_running_loop() + + async def async_result(): + assert asyncio.get_running_loop() is loop + return "coroutine-result" + + def task(*, item): + return async_result() + + assert await _run_task(task, {"input": "question"}) == "coroutine-result" + + +@pytest.mark.asyncio +@pytest.mark.timeout(3) +@pytest.mark.parametrize("cycle", ["wrapped", "partial"]) +async def test_cyclic_wrapper_metadata_keeps_sync_task_execution(cycle): + caller_thread = get_ident() + worker_threads = [] + + def task(*, item): + worker_threads.append(get_ident()) + return item["input"] + + task.__wrapped__ = task if cycle == "wrapped" else partial(task) + + assert await _run_task(task, {"input": "cycle-result"}) == "cycle-result" + assert len(worker_threads) == 1 + assert worker_threads[0] != caller_thread + + class TestRunnerContextDefaults: def test_context_defaults_flow_through(self): ctx_data = [{"input": "a"}] @@ -245,6 +310,172 @@ def _is_optional(annotation) -> bool: class TestExperimentObservationTree: + @pytest.mark.skipif( + not hasattr(inspect, "markcoroutinefunction"), + reason="explicit coroutine markers require Python 3.12+", + ) + @pytest.mark.parametrize("task_form", ["marked", "partial", "wrapped"]) + def test_marked_coroutine_factory_runs_on_experiment_loop( + self, task_form, langfuse_memory_client + ): + def sync_target(*, item, prefix): + return "the wrapped target must not be invoked" + + @wraps(sync_target) + def factory(*, item, prefix="marked"): + asyncio.get_running_loop() + + async def result(): + return f"{prefix}:{item['input']}" + + return result() + + marked = inspect.markcoroutinefunction(factory) + task = marked + expected_prefix = "marked" + if task_form == "partial": + task = partial(marked, prefix="partial") + expected_prefix = "partial" + elif task_form == "wrapped": + + @wraps(marked) + def wrapped(**kwargs): + return marked(**kwargs) + + task = wrapped + + result = langfuse_memory_client.run_experiment( + name="marked-coroutine-factory", + data=[{"input": "question"}], + task=task, + max_concurrency=1, + ) + assert [item.output for item in result.item_results] == [ + f"{expected_prefix}:question" + ] + + @pytest.mark.parametrize( + "task_form,expected_prefix", + [ + ("wrapped", "wrapped"), + ("partial", "partial"), + ("callable", "wrapped"), + ("partial-callable", "partial"), + ], + ) + def test_wrapped_async_task_invokes_wrapper_on_event_loop( + self, task_form, expected_prefix, langfuse_memory_client + ): + wrapper_calls = [] + + async def native_async(*, item, prefix): + return f"{prefix}:{item['input']}" + + @wraps(native_async) + def wrapped(*, item, prefix="wrapped"): + wrapper_calls.append(asyncio.get_running_loop().is_running()) + return native_async(item=item, prefix=prefix) + + result = langfuse_memory_client.run_experiment( + name="wrapped-async-loop", + data=[{"input": "question"}], + task=_task_form(wrapped, task_form), + max_concurrency=1, + ) + + assert [item.output for item in result.item_results] == [ + f"{expected_prefix}:question" + ] + assert wrapper_calls == [True] + + @pytest.mark.parametrize( + "task_form,expected_prefix", + [ + ("wrapped", "wrapped"), + ("partial", "partial"), + ("callable", "wrapped"), + ("partial-callable", "partial"), + ], + ) + def test_wrapped_async_task_preserves_wrapper_otel_context( + self, task_form, expected_prefix, langfuse_memory_client, get_span + ): + key = otel_context.create_key("wrapped-task-context") + + async def read_context(*, item, prefix): + await asyncio.sleep(0) + with langfuse_memory_client.start_as_current_observation( + name="wrapped-async-child" + ): + return f"{prefix}:{otel_context.get_value(key)}" + + @wraps(read_context) + def contextual_wrapper(*, item, prefix="wrapped"): + token = otel_context.attach( + otel_context.set_value(key, "wrapped-otel-context") + ) + + async def finish(): + try: + return await read_context(item=item, prefix=prefix) + finally: + otel_context.detach(token) + + return finish() + + result = langfuse_memory_client.run_experiment( + name="wrapped-async-context", + data=[{"input": "question"}], + task=_task_form(contextual_wrapper, task_form), + max_concurrency=1, + ) + + assert [item.output for item in result.item_results] == [ + f"{expected_prefix}:wrapped-otel-context" + ] + child = get_span("wrapped-async-child") + parent = get_span("experiment-item-task") + assert child.parent.span_id == parent.context.span_id + assert child.context.trace_id == parent.context.trace_id + + def test_sync_tasks_run_concurrently_with_trace_context( + self, langfuse_memory_client, memory_exporter + ): + barrier = Barrier(2, timeout=3) + + def task(*, item): + with langfuse_memory_client.start_as_current_observation( + name=f"worker-{item['input']}" + ): + barrier.wait() + return item["input"].upper() + + result = langfuse_memory_client.run_experiment( + name="sync-task-concurrency", + data=[{"input": "first"}, {"input": "second"}], + task=task, + max_concurrency=2, + ) + + assert [item.output for item in result.item_results] == ["FIRST", "SECOND"] + + spans = memory_exporter.get_finished_spans() + tasks_by_id = { + span.context.span_id: span + for span in spans + if span.name == "experiment-item-task" + } + workers = [span for span in spans if span.name.startswith("worker-")] + assert len(workers) == 2 + assert len({span.parent.span_id for span in workers}) == 2 + for worker in workers: + parent = tasks_by_id[worker.parent.span_id] + assert worker.context.trace_id == parent.context.trace_id + assert ( + worker.attributes[LangfuseOtelSpanAttributes.EXPERIMENT_ID] + == result.experiment_id + ) + def test_failed_task_preserves_experiment_attributes_on_item_run( self, langfuse_memory_client, From 360fc2eb2f5d63a991aaa448ca58df8e287e22eb Mon Sep 17 00:00:00 2001 From: po-et <42566883+po-et@users.noreply.github.com> Date: Tue, 13 Oct 2026 22:20:59 +0800 Subject: [PATCH 7/8] feat(experiments): make synchronous task concurrency opt-in --- MIGRATION_V5_SYNC_TASKS.md | 55 --------- README.md | 4 - langfuse/_client/client.py | 35 +++++- langfuse/_client/datasets.py | 8 ++ langfuse/experiment.py | 40 +++++-- tests/unit/test_experiment.py | 205 +++++++++++++++++++++++++++++++--- 6 files changed, 257 insertions(+), 90 deletions(-) delete mode 100644 MIGRATION_V5_SYNC_TASKS.md diff --git a/MIGRATION_V5_SYNC_TASKS.md b/MIGRATION_V5_SYNC_TASKS.md deleted file mode 100644 index 1c9c787b1..000000000 --- a/MIGRATION_V5_SYNC_TASKS.md +++ /dev/null @@ -1,55 +0,0 @@ -# Synchronous experiment tasks in v5 - -This note describes the execution-context change on the v5 preparation branch; -it does not announce a released SDK version. - -Blocking synchronous experiment tasks now run in worker threads instead of -blocking the experiment's event loop. They can overlap up to the experiment's -`max_concurrency`. Async tasks, including recognized decorated wrappers and -callable objects, continue to run on the experiment's event loop. - -## Audit captured resources before upgrading - -A synchronous task must not assume it runs on the caller's thread. A SQLite -connection created by the caller uses thread checking by default and fails when -used from a worker. The same concern applies to other resources that require a -particular thread. Even `max_concurrency=1` executes the synchronous task on a -worker and does not restore the previous execution context. - -Shared task state may now be accessed concurrently. Use a client that supports -concurrent use, synchronize shared state, or create each task's resource where -it is used. Turning off SQLite's thread check alone does not establish safe -concurrent ownership. - -## Create and close the resource in the synchronous task - -For string dataset inputs, this task opens, uses and closes its connection on -the same worker thread: - -```python -import sqlite3 -from contextlib import closing - - -def task(*, item, **kwargs): - with closing(sqlite3.connect("experiment.sqlite")) as connection: - return connection.execute("SELECT ?", (item["input"],)).fetchone()[0] -``` - -The unit test exercises this pattern through `run_experiment` with two items -and `max_concurrency=2`. - -An alternative is to use an asynchronous client and create/use its resources -on the experiment's event loop. Merely wrapping blocking work in `async def` -can keep that work on the loop and serialize it again. The loop itself is not -guaranteed to be on the original caller's thread, including notebook use. - -## Inspect failed items as well as successful outputs - -The existing item-error policy is unchanged: task exceptions are logged and -recorded in the trace, and failed items do not appear in `item_results`. -`run_experiment` does not re-raise each task's exception. A plain script that -captures a caller-owned SQLite connection can therefore return zero item -results without raising that SQLite exception. Check the logged failures and -returned items during migration; successful completion of the runner alone -does not prove that every task succeeded. diff --git a/README.md b/README.md index 4f3246477..35b3bb5e7 100644 --- a/README.md +++ b/README.md @@ -22,10 +22,6 @@ The [Langfuse](https://langfuse.com) Python SDK covers the full platform: **observability/tracing** (OpenTelemetry-based, with OpenAI and LangChain integrations), **datasets & experiments** (offline evaluation and regression testing of prompt/model changes, including [CI via GitHub Actions](https://github.com/langfuse/experiment-action)), **LLM-as-a-judge and custom evaluations/scores**, **prompt management**, and a **full REST API client**. -On the v5 preparation branch, synchronous experiment tasks run in worker threads. -This changes resource ownership and concurrency requirements; see the -[synchronous task migration notes](MIGRATION_V5_SYNC_TASKS.md) before upgrading. - ## Installation ``` diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index d8349f802..b52fcb0ae 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -9,6 +9,7 @@ import re import urllib.parse import uuid +from concurrent.futures import ThreadPoolExecutor from datetime import datetime from hashlib import sha256 from time import time_ns @@ -2461,6 +2462,7 @@ def run_experiment( composite_evaluator: Optional[CompositeEvaluatorFunction] = None, run_evaluators: List[RunEvaluatorFunction] = [], max_concurrency: int = 50, + run_sync_tasks_in_threads: bool = False, metadata: Optional[Dict[str, str]] = None, _dataset_version: Optional[datetime] = None, ) -> ExperimentResult: @@ -2506,6 +2508,17 @@ def run_experiment( max_concurrency: Maximum number of concurrent task executions (default: 50). Controls the number of items processed simultaneously. Adjust based on API rate limits and system resources. + run_sync_tasks_in_threads: Opt in to running synchronous tasks in a + dedicated thread pool with up to `max_concurrency` workers (default: + False). This enables concurrency for blocking I/O without changing + the default event-loop execution. Async tasks stay on the experiment + loop. When enabled, tasks must be thread-safe: create and close + thread-bound resources, such as SQLite connections, inside the task. + Even `max_concurrency=1` offloads synchronous tasks. Exceptions are + still logged and failed items omitted. Cancellation does not stop + tasks already running in threads, which can continue after the + experiment returns and delay Python process exit. Use finite I/O + timeouts. For cooperative cancellation, prefer async tasks. metadata: Optional metadata dictionary to attach to all experiment traces. This metadata will be included in every trace created during the experiment and is shown as the experiment's metadata in Langfuse. @@ -2648,6 +2661,7 @@ def average_accuracy(*, item_results, **kwargs): composite_evaluator=composite_evaluator, run_evaluators=run_evaluators or [], max_concurrency=max_concurrency, + run_sync_tasks_in_threads=run_sync_tasks_in_threads, metadata=metadata, dataset_version=_dataset_version, ), @@ -2666,6 +2680,7 @@ async def _run_experiment_async( composite_evaluator: Optional[CompositeEvaluatorFunction], run_evaluators: List[RunEvaluatorFunction], max_concurrency: int, + run_sync_tasks_in_threads: bool = False, metadata: Optional[Dict[str, Any]] = None, dataset_version: Optional[datetime] = None, ) -> ExperimentResult: @@ -2704,6 +2719,11 @@ async def _run_experiment_async( # Set up concurrency control semaphore = asyncio.Semaphore(max_concurrency) + task_executor = ( + ThreadPoolExecutor(max_workers=max_concurrency) + if run_sync_tasks_in_threads + else None + ) # Process all items async def process_item(item: ExperimentItem) -> ExperimentItemResult: @@ -2719,11 +2739,18 @@ async def process_item(item: ExperimentItem) -> ExperimentItemResult: description, metadata, experiment_item_version, + task_executor=task_executor, ) # Run all items concurrently - tasks = [process_item(item) for item in data] - item_results = await asyncio.gather(*tasks, return_exceptions=True) + try: + tasks = [process_item(item) for item in data] + item_results = await asyncio.gather(*tasks, return_exceptions=True) + finally: + if task_executor is not None: + # Do not block the event loop on cancellation. Running calls + # cannot be stopped, but queued calls must not start afterward. + task_executor.shutdown(wait=False, cancel_futures=True) # Filter out any exceptions and log errors valid_results: List[ExperimentItemResult] = [] @@ -2860,6 +2887,8 @@ async def _process_experiment_item( experiment_description: Optional[str], experiment_metadata: Optional[Dict[str, Any]] = None, experiment_item_version: Optional[str] = None, + *, + task_executor: Optional[ThreadPoolExecutor] = None, ) -> ExperimentItemResult: with self.start_as_current_observation(name="experiment-item-run") as span: try: @@ -2973,7 +3002,7 @@ async def _process_experiment_item( # metadata instead of leaving no room for the output. task_span.update(metadata=final_observation_metadata) try: - output = await _run_task(task, item) + output = await _run_task(task, item, executor=task_executor) except Exception as e: task_span.update( output=f"Error: {str(e)}", diff --git a/langfuse/_client/datasets.py b/langfuse/_client/datasets.py index 1070c76d3..8a9a25296 100644 --- a/langfuse/_client/datasets.py +++ b/langfuse/_client/datasets.py @@ -89,6 +89,7 @@ def run_experiment( composite_evaluator: Optional[CompositeEvaluatorFunction] = None, run_evaluators: List[RunEvaluatorFunction] = [], max_concurrency: int = 50, + run_sync_tasks_in_threads: bool = False, metadata: Optional[Dict[str, Any]] = None, ) -> ExperimentResult: """Run an experiment on this Langfuse dataset with automatic tracking. @@ -126,6 +127,12 @@ def run_experiment( Useful for computing aggregate statistics across all dataset items. max_concurrency: Maximum number of concurrent task executions (default: 50). Adjust based on API rate limits and system resources. + run_sync_tasks_in_threads: Opt in to concurrent synchronous tasks in a + dedicated pool with up to `max_concurrency` workers (default: False). + Tasks must be thread-safe; create thread-bound resources inside the + task. Async tasks stay on the experiment loop. Cancellation cannot + stop synchronous tasks already running in threads. See + `Langfuse.run_experiment` for resource ownership and timeout guidance. metadata: Optional metadata to attach to the experiment run and all traces. Will be combined with individual item metadata. @@ -300,6 +307,7 @@ def content_diversity(*, item_results, **kwargs): composite_evaluator=composite_evaluator, run_evaluators=run_evaluators, max_concurrency=max_concurrency, + run_sync_tasks_in_threads=run_sync_tasks_in_threads, metadata=metadata, _dataset_version=self.version, ) diff --git a/langfuse/experiment.py b/langfuse/experiment.py index c4a937f24..ff5829296 100644 --- a/langfuse/experiment.py +++ b/langfuse/experiment.py @@ -8,6 +8,8 @@ import asyncio import inspect import warnings +from concurrent.futures import ThreadPoolExecutor +from contextvars import copy_context from datetime import datetime from functools import partial from typing import ( @@ -645,15 +647,17 @@ class TaskFunction(Protocol): - Can be either synchronous or asynchronous - Should handle their own errors gracefully (exceptions will be logged) - Synchronous task functions run in worker threads so blocking calls can run - concurrently. Asynchronous task functions, including their decorated wrappers, - run on the experiment's event loop. - - This is a breaking execution-context change in v5. Do not capture a resource - that can only be used on the caller's thread, such as a default sqlite3 - connection, in a synchronous task. Create and close the resource inside the - task or use an asynchronous client on the experiment's event loop. Setting - max_concurrency to 1 does not restore caller-thread execution. + By default, synchronous task functions run on the experiment's event loop, + so blocking calls execute serially. Set `run_sync_tasks_in_threads=True` on + `run_experiment` to run synchronous tasks concurrently in worker threads. + Asynchronous tasks, including decorated wrappers, stay on the event loop. + + When opting in, synchronous tasks must be thread-safe. Create and close + thread-bound resources such as sqlite3 connections inside the task, rather + than capturing a connection created on the caller's thread. Even + `max_concurrency=1` offloads the task. Task exceptions are logged and failed + items are omitted from the result. Cancellation cannot stop a synchronous + task already running in a thread; use finite timeouts on blocking I/O. """ def __call__( @@ -1095,13 +1099,23 @@ def _is_async_task(task: Any) -> bool: return False -async def _run_task(task: TaskFunction, item: ExperimentItem) -> Any: +async def _run_task( + task: TaskFunction, + item: ExperimentItem, + *, + executor: Optional[ThreadPoolExecutor] = None, +) -> Any: """Run a task function and handle sync/async.""" result: Any - if _is_async_task(task): + if executor is None or _is_async_task(task): result = task(item=item) else: - result = await asyncio.to_thread(task, item=item) + # run_in_executor does not propagate ContextVars itself. Copy separately + # for every item so worker spans inherit that item's tracing context. + context = copy_context() + result = await asyncio.get_running_loop().run_in_executor( + executor, context.run, partial(task, item=item) + ) # Handle async tasks if asyncio.iscoroutine(result): @@ -1206,6 +1220,7 @@ def run_experiment( composite_evaluator: Optional["CompositeEvaluatorFunction"] = None, run_evaluators: List[RunEvaluatorFunction] = [], max_concurrency: int = 50, + run_sync_tasks_in_threads: bool = False, metadata: Optional[Dict[str, str]] = None, _dataset_version: Optional[datetime] = None, ) -> ExperimentResult: @@ -1235,6 +1250,7 @@ def run_experiment( composite_evaluator=composite_evaluator, run_evaluators=run_evaluators, max_concurrency=max_concurrency, + run_sync_tasks_in_threads=run_sync_tasks_in_threads, metadata=merged_metadata, _dataset_version=resolved_dataset_version, ) diff --git a/tests/unit/test_experiment.py b/tests/unit/test_experiment.py index 995d98e67..c66fcf6d5 100644 --- a/tests/unit/test_experiment.py +++ b/tests/unit/test_experiment.py @@ -5,10 +5,11 @@ import json import sqlite3 import typing +from concurrent.futures import ThreadPoolExecutor from contextlib import closing from datetime import datetime, timezone from functools import partial, wraps -from threading import Barrier, get_ident +from threading import Barrier, Event, get_ident from typing import get_type_hints from unittest.mock import MagicMock @@ -19,7 +20,8 @@ from langfuse import Evaluation, RegressionError, RunnerContext from langfuse._client.attributes import LangfuseOtelSpanAttributes from langfuse._client.client import Langfuse -from langfuse.api import DatasetItem, DatasetStatus +from langfuse._client.datasets import DatasetClient +from langfuse.api import Dataset, DatasetItem, DatasetStatus from langfuse.batch_evaluation import CompositeEvaluatorFunction from langfuse.experiment import ExperimentItemResult, ExperimentResult, _run_task @@ -51,19 +53,68 @@ def __call__(self, **kwargs): return task +def _experiment_runner(client, entrypoint, inputs): + if entrypoint == "client": + return partial(client.run_experiment, data=inputs) + if entrypoint == "context": + return RunnerContext(client=client, data=inputs).run_experiment + + timestamp = datetime(2026, 1, 1, tzinfo=timezone.utc) + dataset = DatasetClient( + dataset=Dataset( + id="task-dataset", + name="task-dataset", + project_id="project", + metadata=None, + created_at=timestamp, + updated_at=timestamp, + ), + items=[ + DatasetItem( + id=str(i), + input=item["input"], + expected_output=None, + metadata=None, + media_references=[], + status=DatasetStatus.ACTIVE, + dataset_id="task-dataset", + dataset_name="task-dataset", + created_at=timestamp, + updated_at=timestamp, + ) + for i, item in enumerate(inputs) + ], + langfuse_client=client, + ) + return dataset.run_experiment + + +@pytest.fixture +def task_executor(): + with ThreadPoolExecutor(max_workers=2) as executor: + yield executor + + @pytest.mark.asyncio -async def test_async_task_stays_on_experiment_event_loop(): +@pytest.mark.parametrize("offload", [False, True]) +async def test_async_task_stays_on_experiment_event_loop(offload, task_executor): loop = asyncio.get_running_loop() async def task(*, item): assert asyncio.get_running_loop() is loop return item["input"] - assert await _run_task(task, {"input": "async-result"}) == "async-result" + assert ( + await _run_task( + task, {"input": "async-result"}, executor=task_executor if offload else None + ) + == "async-result" + ) @pytest.mark.asyncio -async def test_sync_task_returning_coroutine_is_awaited(): +@pytest.mark.parametrize("offload", [False, True]) +async def test_sync_task_returning_coroutine_is_awaited(offload, task_executor): loop = asyncio.get_running_loop() async def async_result(): @@ -73,13 +124,18 @@ async def async_result(): def task(*, item): return async_result() - assert await _run_task(task, {"input": "question"}) == "coroutine-result" + assert ( + await _run_task( + task, {"input": "question"}, executor=task_executor if offload else None + ) + == "coroutine-result" + ) @pytest.mark.asyncio @pytest.mark.timeout(3) @pytest.mark.parametrize("cycle", ["wrapped", "partial"]) -async def test_cyclic_wrapper_metadata_keeps_sync_task_execution(cycle): +async def test_cyclic_wrapper_metadata_keeps_sync_task_execution(cycle, task_executor): caller_thread = get_ident() worker_threads = [] @@ -89,7 +145,10 @@ def task(*, item): task.__wrapped__ = task if cycle == "wrapped" else partial(task) - assert await _run_task(task, {"input": "cycle-result"}) == "cycle-result" + assert ( + await _run_task(task, {"input": "cycle-result"}, executor=task_executor) + == "cycle-result" + ) assert len(worker_threads) == 1 assert worker_threads[0] != caller_thread @@ -314,6 +373,102 @@ def _is_optional(annotation) -> bool: class TestExperimentObservationTree: + @pytest.mark.parametrize("entrypoint", ["client", "context", "dataset"]) + @pytest.mark.parametrize("flag", [None, False]) + def test_default_sync_task_preserves_caller_owned_sqlite_connection( + self, entrypoint, flag, langfuse_memory_client + ): + inputs = [{"input": "A"}, {"input": "B"}] + run_experiment = _experiment_runner(langfuse_memory_client, entrypoint, inputs) + kwargs = {} + if flag is not None: + kwargs["run_sync_tasks_in_threads"] = flag + + caller_thread = get_ident() + with closing(sqlite3.connect(":memory:")) as connection: + + def task(*, item): + assert get_ident() == caller_thread + value = item["input"] if isinstance(item, dict) else item.input + return connection.execute("SELECT ?", (value,)).fetchone()[0] + + result = run_experiment( + name="default-sync-task", + task=task, + max_concurrency=2, + **kwargs, + ) + + assert [item.output for item in result.item_results] == ["A", "B"] + + @pytest.mark.asyncio + async def test_opt_in_sync_tasks_use_experiment_concurrency_not_default_pool( + self, langfuse_memory_client + ): + loop = asyncio.get_running_loop() + loop.set_default_executor(ThreadPoolExecutor(max_workers=1)) + barrier = Barrier(4, timeout=3) + + def task(*, item): + barrier.wait() + return item["input"] + + result = await langfuse_memory_client._run_experiment_async( + name="dedicated-sync-task-pool", + run_name="dedicated-sync-task-pool", + description=None, + data=[{"input": i} for i in range(4)], + task=task, + evaluators=[], + composite_evaluator=None, + run_evaluators=[], + max_concurrency=4, + run_sync_tasks_in_threads=True, + ) + + assert [item.output for item in result.item_results] == [0, 1, 2, 3] + + @pytest.mark.asyncio + async def test_cancelling_experiment_does_not_wait_for_running_sync_task( + self, langfuse_memory_client + ): + loop = asyncio.get_running_loop() + started = asyncio.Event() + release = Event() + finished = Event() + + def task(*, item): + loop.call_soon_threadsafe(started.set) + try: + release.wait(timeout=3) + return item["input"] + finally: + finished.set() + + experiment = asyncio.create_task( + langfuse_memory_client._run_experiment_async( + name="cancel-sync-task", + run_name="cancel-sync-task", + description=None, + data=[{"input": "A"}], + task=task, + evaluators=[], + composite_evaluator=None, + run_evaluators=[], + max_concurrency=1, + run_sync_tasks_in_threads=True, + ) + ) + try: + await asyncio.wait_for(started.wait(), timeout=2) + experiment.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(experiment, timeout=1) + assert not finished.is_set() + finally: + release.set() + await asyncio.to_thread(finished.wait, 2) + def test_thread_affine_sync_task_records_item_failures( self, langfuse_memory_client, caplog ): @@ -327,6 +482,7 @@ def task(*, item, **kwargs): data=[{"input": "A"}, {"input": "B"}], task=task, max_concurrency=2, + run_sync_tasks_in_threads=True, ) assert result.item_results == [] @@ -348,6 +504,7 @@ def task(*, item, **kwargs): data=[{"input": "A"}, {"input": "B"}], task=task, max_concurrency=2, + run_sync_tasks_in_threads=True, ) assert [item.output for item in result.item_results] == ["A", "B"] @@ -357,8 +514,9 @@ def task(*, item, **kwargs): reason="explicit coroutine markers require Python 3.12+", ) @pytest.mark.parametrize("task_form", ["marked", "partial", "wrapped"]) + @pytest.mark.parametrize("offload", [False, True]) def test_marked_coroutine_factory_runs_on_experiment_loop( - self, task_form, langfuse_memory_client + self, task_form, offload, langfuse_memory_client ): def sync_target(*, item, prefix): return "the wrapped target must not be invoked" @@ -391,6 +549,7 @@ def wrapped(**kwargs): data=[{"input": "question"}], task=task, max_concurrency=1, + run_sync_tasks_in_threads=offload, ) assert [item.output for item in result.item_results] == [ f"{expected_prefix}:question" @@ -405,8 +564,9 @@ def wrapped(**kwargs): ("partial-callable", "partial"), ], ) + @pytest.mark.parametrize("offload", [False, True]) def test_wrapped_async_task_invokes_wrapper_on_event_loop( - self, task_form, expected_prefix, langfuse_memory_client + self, task_form, expected_prefix, offload, langfuse_memory_client ): wrapper_calls = [] @@ -423,6 +583,7 @@ def wrapped(*, item, prefix="wrapped"): data=[{"input": "question"}], task=_task_form(wrapped, task_form), max_concurrency=1, + run_sync_tasks_in_threads=offload, ) assert [item.output for item in result.item_results] == [ @@ -439,8 +600,9 @@ def wrapped(*, item, prefix="wrapped"): ("partial-callable", "partial"), ], ) + @pytest.mark.parametrize("offload", [False, True]) def test_wrapped_async_task_preserves_wrapper_otel_context( - self, task_form, expected_prefix, langfuse_memory_client, get_span + self, task_form, expected_prefix, offload, langfuse_memory_client, get_span ): key = otel_context.create_key("wrapped-task-context") @@ -470,6 +632,7 @@ async def finish(): data=[{"input": "question"}], task=_task_form(contextual_wrapper, task_form), max_concurrency=1, + run_sync_tasks_in_threads=offload, ) assert [item.output for item in result.item_results] == [ @@ -480,23 +643,30 @@ async def finish(): assert child.parent.span_id == parent.context.span_id assert child.context.trace_id == parent.context.trace_id + @pytest.mark.parametrize("entrypoint", ["client", "context", "dataset"]) def test_sync_tasks_run_concurrently_with_trace_context( - self, langfuse_memory_client, memory_exporter + self, entrypoint, langfuse_memory_client, memory_exporter ): barrier = Barrier(2, timeout=3) def task(*, item): + value = item["input"] if isinstance(item, dict) else item.input with langfuse_memory_client.start_as_current_observation( - name=f"worker-{item['input']}" + name=f"worker-{value}" ): barrier.wait() - return item["input"].upper() + return value.upper() - result = langfuse_memory_client.run_experiment( + run_experiment = _experiment_runner( + langfuse_memory_client, + entrypoint, + [{"input": "first"}, {"input": "second"}], + ) + result = run_experiment( name="sync-task-concurrency", - data=[{"input": "first"}, {"input": "second"}], task=task, max_concurrency=2, + run_sync_tasks_in_threads=True, ) assert [item.output for item in result.item_results] == ["FIRST", "SECOND"] @@ -518,8 +688,10 @@ def task(*, item): == result.experiment_id ) + @pytest.mark.parametrize("offload", [False, True]) def test_failed_task_preserves_experiment_attributes_on_item_run( self, + offload, langfuse_memory_client, get_span, ): @@ -532,6 +704,7 @@ def failing_task(**kwargs): task=failing_task, metadata={"run": "metadata"}, max_concurrency=1, + run_sync_tasks_in_threads=offload, ) item_run = get_span("experiment-item-run") From 4a2a263068ec246b9c24fcb3016ba862bc12fef8 Mon Sep 17 00:00:00 2001 From: po-et <42566883+po-et@users.noreply.github.com> Date: Tue, 13 Oct 2026 22:35:47 +0800 Subject: [PATCH 8/8] fix(experiments): close cancelled task coroutine results --- langfuse/experiment.py | 23 ++++++++++++++++------ tests/unit/test_experiment.py | 36 +++++++++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 6 deletions(-) diff --git a/langfuse/experiment.py b/langfuse/experiment.py index ff5829296..65986d1b7 100644 --- a/langfuse/experiment.py +++ b/langfuse/experiment.py @@ -8,7 +8,7 @@ import asyncio import inspect import warnings -from concurrent.futures import ThreadPoolExecutor +from concurrent.futures import Future, ThreadPoolExecutor from contextvars import copy_context from datetime import datetime from functools import partial @@ -1110,12 +1110,23 @@ async def _run_task( if executor is None or _is_async_task(task): result = task(item=item) else: - # run_in_executor does not propagate ContextVars itself. Copy separately - # for every item so worker spans inherit that item's tracing context. + # Copy separately for every item so worker spans inherit that item's + # tracing context. Keep the source future to dispose of a late coroutine + # if cancellation prevents the caller from awaiting it. context = copy_context() - result = await asyncio.get_running_loop().run_in_executor( - executor, context.run, partial(task, item=item) - ) + future = executor.submit(context.run, partial(task, item=item)) + try: + result = await asyncio.wrap_future(future) + except asyncio.CancelledError: + + def discard_result(done: Future[Any]) -> None: + if not done.cancelled() and done.exception() is None: + value = done.result() + if asyncio.iscoroutine(value): + value.close() + + future.add_done_callback(discard_result) + raise # Handle async tasks if asyncio.iscoroutine(result): diff --git a/tests/unit/test_experiment.py b/tests/unit/test_experiment.py index c66fcf6d5..5f09d30fd 100644 --- a/tests/unit/test_experiment.py +++ b/tests/unit/test_experiment.py @@ -132,6 +132,42 @@ def task(*, item): ) +@pytest.mark.asyncio +async def test_cancelled_sync_factory_closes_its_unawaited_coroutine(task_executor): + loop = asyncio.get_running_loop() + started = asyncio.Event() + release = Event() + results = [] + + async def result(): + return "cancelled-result" + + def task(*, item): + loop.call_soon_threadsafe(started.set) + release.wait(timeout=3) + coroutine = result() + results.append(coroutine) + return coroutine + + running = asyncio.create_task( + _run_task(task, {"input": "A"}, executor=task_executor) + ) + try: + await asyncio.wait_for(started.wait(), timeout=2) + running.cancel() + with pytest.raises(asyncio.CancelledError): + await running + release.set() + await asyncio.to_thread(task_executor.shutdown, wait=True) + assert len(results) == 1 + assert inspect.getcoroutinestate(results[0]) == inspect.CORO_CLOSED + finally: + release.set() + await asyncio.to_thread(task_executor.shutdown, wait=True) + for coroutine in results: + coroutine.close() + + @pytest.mark.asyncio @pytest.mark.timeout(3) @pytest.mark.parametrize("cycle", ["wrapped", "partial"])