langfuse.types

@private

 1"""@private"""
 2
 3from datetime import datetime
 4from langfuse.client import PromptClient, ModelUsage, MapValue
 5from typing import Any, List, Optional, TypedDict, Literal, Dict, Union, Protocol
 6from pydantic import BaseModel
 7
 8SpanLevel = Literal["DEBUG", "DEFAULT", "WARNING", "ERROR"]
 9
10ScoreDataType = Literal["NUMERIC", "CATEGORICAL", "BOOLEAN"]
11
12
13class TraceMetadata(TypedDict):
14    name: Optional[str]
15    user_id: Optional[str]
16    session_id: Optional[str]
17    version: Optional[str]
18    release: Optional[str]
19    metadata: Optional[Any]
20    tags: Optional[List[str]]
21    public: Optional[bool]
22
23
24class ObservationParams(TraceMetadata, TypedDict):
25    input: Optional[Any]
26    output: Optional[Any]
27    level: Optional[SpanLevel]
28    status_message: Optional[str]
29    start_time: Optional[datetime]
30    end_time: Optional[datetime]
31    completion_start_time: Optional[datetime]
32    model: Optional[str]
33    model_parameters: Optional[Dict[str, MapValue]]
34    usage: Optional[Union[BaseModel, ModelUsage]]
35    prompt: Optional[PromptClient]
36
37
38class MaskFunction(Protocol):
39    """A function that masks data.
40
41    Keyword Args:
42        data: The data to mask.
43
44    Returns:
45        The masked data that must be serializable to JSON.
46    """
47
48    def __call__(self, *, data: Any) -> Any: ...
SpanLevel = typing.Literal['DEBUG', 'DEFAULT', 'WARNING', 'ERROR']
ScoreDataType = typing.Literal['NUMERIC', 'CATEGORICAL', 'BOOLEAN']
class TraceMetadata(typing.TypedDict):
14class TraceMetadata(TypedDict):
15    name: Optional[str]
16    user_id: Optional[str]
17    session_id: Optional[str]
18    version: Optional[str]
19    release: Optional[str]
20    metadata: Optional[Any]
21    tags: Optional[List[str]]
22    public: Optional[bool]
name: Optional[str]
user_id: Optional[str]
session_id: Optional[str]
version: Optional[str]
release: Optional[str]
metadata: Optional[Any]
tags: Optional[List[str]]
public: Optional[bool]
class ObservationParams(TraceMetadata, typing.TypedDict):
25class ObservationParams(TraceMetadata, TypedDict):
26    input: Optional[Any]
27    output: Optional[Any]
28    level: Optional[SpanLevel]
29    status_message: Optional[str]
30    start_time: Optional[datetime]
31    end_time: Optional[datetime]
32    completion_start_time: Optional[datetime]
33    model: Optional[str]
34    model_parameters: Optional[Dict[str, MapValue]]
35    usage: Optional[Union[BaseModel, ModelUsage]]
36    prompt: Optional[PromptClient]
input: Optional[Any]
output: Optional[Any]
level: Optional[Literal['DEBUG', 'DEFAULT', 'WARNING', 'ERROR']]
status_message: Optional[str]
start_time: Optional[datetime.datetime]
end_time: Optional[datetime.datetime]
completion_start_time: Optional[datetime.datetime]
model: Optional[str]
model_parameters: Optional[Dict[str, Union[str, NoneType, int, bool, List[str]]]]
usage: Union[pydantic.main.BaseModel, langfuse.model.ModelUsage, NoneType]
class MaskFunction(typing.Protocol):
39class MaskFunction(Protocol):
40    """A function that masks data.
41
42    Keyword Args:
43        data: The data to mask.
44
45    Returns:
46        The masked data that must be serializable to JSON.
47    """
48
49    def __call__(self, *, data: Any) -> Any: ...

A function that masks data.

Keyword Args:

data: The data to mask.

Returns:

The masked data that must be serializable to JSON.

MaskFunction(*args, **kwargs)
1927def _no_init_or_replace_init(self, *args, **kwargs):
1928    cls = type(self)
1929
1930    if cls._is_protocol:
1931        raise TypeError('Protocols cannot be instantiated')
1932
1933    # Already using a custom `__init__`. No need to calculate correct
1934    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1935    if cls.__init__ is not _no_init_or_replace_init:
1936        return
1937
1938    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1939    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1940    # searches for a proper new `__init__` in the MRO. The new `__init__`
1941    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1942    # instantiation of the protocol subclass will thus use the new
1943    # `__init__` and no longer call `_no_init_or_replace_init`.
1944    for base in cls.__mro__:
1945        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1946        if init is not _no_init_or_replace_init:
1947            cls.__init__ = init
1948            break
1949    else:
1950        # should not happen
1951        cls.__init__ = object.__init__
1952
1953    cls.__init__(self, *args, **kwargs)