langfuse.types

@private

 1"""@private"""
 2
 3from datetime import datetime
 4from typing import Any, Dict, List, Literal, Optional, Protocol, TypedDict, Union
 5
 6from pydantic import BaseModel
 7from langfuse.api import MediaContentType
 8from langfuse.model import MapValue, ModelUsage, PromptClient
 9
10SpanLevel = Literal["DEBUG", "DEFAULT", "WARNING", "ERROR"]
11
12ScoreDataType = Literal["NUMERIC", "CATEGORICAL", "BOOLEAN"]
13
14
15class TraceMetadata(TypedDict):
16    name: Optional[str]
17    user_id: Optional[str]
18    session_id: Optional[str]
19    version: Optional[str]
20    release: Optional[str]
21    metadata: Optional[Any]
22    tags: Optional[List[str]]
23    public: Optional[bool]
24
25
26class ObservationParams(TraceMetadata, TypedDict):
27    input: Optional[Any]
28    output: Optional[Any]
29    level: Optional[SpanLevel]
30    status_message: Optional[str]
31    start_time: Optional[datetime]
32    end_time: Optional[datetime]
33    completion_start_time: Optional[datetime]
34    model: Optional[str]
35    model_parameters: Optional[Dict[str, MapValue]]
36    usage: Optional[Union[BaseModel, ModelUsage]]
37    prompt: Optional[PromptClient]
38
39
40class MaskFunction(Protocol):
41    """A function that masks data.
42
43    Keyword Args:
44        data: The data to mask.
45
46    Returns:
47        The masked data that must be serializable to JSON.
48    """
49
50    def __call__(self, *, data: Any) -> Any: ...
51
52
53class ParsedMediaReference(TypedDict):
54    """A parsed media reference.
55
56    Attributes:
57        media_id: The media ID.
58        source: The original source of the media, e.g. a file path, bytes, base64 data URI, etc.
59        content_type: The content type of the media.
60    """
61
62    media_id: str
63    source: str
64    content_type: MediaContentType
SpanLevel = typing.Literal['DEBUG', 'DEFAULT', 'WARNING', 'ERROR']
ScoreDataType = typing.Literal['NUMERIC', 'CATEGORICAL', 'BOOLEAN']
class TraceMetadata(typing.TypedDict): 
16class TraceMetadata(TypedDict):
17    name: Optional[str]
18    user_id: Optional[str]
19    session_id: Optional[str]
20    version: Optional[str]
21    release: Optional[str]
22    metadata: Optional[Any]
23    tags: Optional[List[str]]
24    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): 
27class ObservationParams(TraceMetadata, TypedDict):
28    input: Optional[Any]
29    output: Optional[Any]
30    level: Optional[SpanLevel]
31    status_message: Optional[str]
32    start_time: Optional[datetime]
33    end_time: Optional[datetime]
34    completion_start_time: Optional[datetime]
35    model: Optional[str]
36    model_parameters: Optional[Dict[str, MapValue]]
37    usage: Optional[Union[BaseModel, ModelUsage]]
38    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]
Inherited Members
TraceMetadata
name
user_id
session_id
version
release
metadata
tags
public
class MaskFunction(typing.Protocol): 
41class MaskFunction(Protocol):
42    """A function that masks data.
43
44    Keyword Args:
45        data: The data to mask.
46
47    Returns:
48        The masked data that must be serializable to JSON.
49    """
50
51    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)
class ParsedMediaReference(typing.TypedDict): 
54class ParsedMediaReference(TypedDict):
55    """A parsed media reference.
56
57    Attributes:
58        media_id: The media ID.
59        source: The original source of the media, e.g. a file path, bytes, base64 data URI, etc.
60        content_type: The content type of the media.
61    """
62
63    media_id: str
64    source: str
65    content_type: MediaContentType

A parsed media reference.

Attributes:
  • media_id: The media ID.
  • source: The original source of the media, e.g. a file path, bytes, base64 data URI, etc.
  • content_type: The content type of the media.
media_id: str
source: str
content_type: Literal['image/png', 'image/jpeg', 'image/jpg', 'image/webp', 'audio/mpeg', 'audio/mp3', 'audio/wav', 'text/plain', 'application/pdf']