1import datetime as dt
   2import logging
   3import os
   4import time
   5import tracemalloc
   6import typing
   7import urllib.parse
   8import uuid
   9import warnings
  10from contextlib import contextmanager
  11from dataclasses import dataclass
  12from enum import Enum
  13from typing import Any, Dict, List, Literal, Optional, Sequence, Union, overload
  14
  15import backoff
  16import httpx
  17
  18from langfuse.api.resources.commons.types.dataset_run_with_items import (
  19    DatasetRunWithItems,
  20)
  21from langfuse.api.resources.commons.types.observations_view import ObservationsView
  22from langfuse.api.resources.commons.types.session import Session
  23from langfuse.api.resources.commons.types.trace_with_details import TraceWithDetails
  24from langfuse.api.resources.datasets.types.paginated_dataset_runs import (
  25    PaginatedDatasetRuns,
  26)
  27from langfuse.api.resources.ingestion.types.create_event_body import CreateEventBody
  28from langfuse.api.resources.ingestion.types.create_generation_body import (
  29    CreateGenerationBody,
  30)
  31from langfuse.api.resources.ingestion.types.create_span_body import CreateSpanBody
  32from langfuse.api.resources.ingestion.types.score_body import ScoreBody
  33from langfuse.api.resources.ingestion.types.sdk_log_body import SdkLogBody
  34from langfuse.api.resources.ingestion.types.trace_body import TraceBody
  35from langfuse.api.resources.ingestion.types.update_generation_body import (
  36    UpdateGenerationBody,
  37)
  38from langfuse.api.resources.ingestion.types.update_span_body import UpdateSpanBody
  39from langfuse.api.resources.media import GetMediaResponse
  40from langfuse.api.resources.observations.types.observations_views import (
  41    ObservationsViews,
  42)
  43from langfuse.api.resources.prompts.types import (
  44    CreatePromptRequest_Chat,
  45    CreatePromptRequest_Text,
  46    Prompt_Chat,
  47    Prompt_Text,
  48)
  49from langfuse.api.resources.trace.types.traces import Traces
  50from langfuse.api.resources.utils.resources.pagination.types.meta_response import (
  51    MetaResponse,
  52)
  53from langfuse.model import (
  54    ChatMessageDict,
  55    ChatPromptClient,
  56    CreateDatasetItemRequest,
  57    CreateDatasetRequest,
  58    CreateDatasetRunItemRequest,
  59    DatasetItem,
  60    DatasetStatus,
  61    ModelUsage,
  62    PromptClient,
  63    TextPromptClient,
  64)
  65from langfuse.parse_error import handle_fern_exception
  66from langfuse.prompt_cache import PromptCache
  67
  68try:
  69    import pydantic.v1 as pydantic  # type: ignore
  70except ImportError:
  71    import pydantic  # type: ignore
  72
  73from langfuse._task_manager.task_manager import TaskManager
  74from langfuse.api.client import AsyncFernLangfuse, FernLangfuse
  75from langfuse.environment import get_common_release_envs
  76from langfuse.logging import clean_logger
  77from langfuse.media import LangfuseMedia
  78from langfuse.model import Dataset, MapValue, Observation, TraceWithFullDetails
  79from langfuse.request import LangfuseClient
  80from langfuse.types import MaskFunction, ScoreDataType, SpanLevel
  81from langfuse.utils import (
  82    _convert_usage_input,
  83    _create_prompt_context,
  84    _get_timestamp,
  85)
  86
  87from .version import __version__ as version
  88
  89
  90@dataclass
  91class FetchTracesResponse:
  92    """Response object for fetch_traces method."""
  93
  94    data: typing.List[TraceWithDetails]
  95    meta: MetaResponse
  96
  97
  98@dataclass
  99class FetchTraceResponse:
 100    """Response object for fetch_trace method."""
 101
 102    data: TraceWithFullDetails
 103
 104
 105@dataclass
 106class FetchObservationsResponse:
 107    """Response object for fetch_observations method."""
 108
 109    data: typing.List[ObservationsView]
 110    meta: MetaResponse
 111
 112
 113@dataclass
 114class FetchObservationResponse:
 115    """Response object for fetch_observation method."""
 116
 117    data: Observation
 118
 119
 120@dataclass
 121class FetchMediaResponse:
 122    """Response object for fetch_media method."""
 123
 124    data: GetMediaResponse
 125
 126
 127@dataclass
 128class FetchSessionsResponse:
 129    """Response object for fetch_sessions method."""
 130
 131    data: typing.List[Session]
 132    meta: MetaResponse
 133
 134
 135class Langfuse(object):
 136    """Langfuse Python client.
 137
 138    Attributes:
 139        log (logging.Logger): Logger for the Langfuse client.
 140        base_url (str): Base URL of the Langfuse API, serving as the root address for API endpoint construction.
 141        httpx_client (httpx.Client): HTTPX client utilized for executing requests to the Langfuse API.
 142        client (FernLangfuse): Core interface for Langfuse API interaction.
 143        task_manager (TaskManager): Task Manager dedicated to handling asynchronous tasks.
 144        release (str): Identifies the release number or hash of the application.
 145        prompt_cache (PromptCache): A cache for efficiently storing and retrieving PromptClient instances.
 146
 147    Example:
 148        Initiating the Langfuse client should always be first step to use Langfuse.
 149        ```python
 150        import os
 151        from langfuse import Langfuse
 152
 153        # Set the public and secret keys as environment variables
 154        os.environ['LANGFUSE_PUBLIC_KEY'] = public_key
 155        os.environ['LANGFUSE_SECRET_KEY'] = secret_key
 156
 157        # Initialize the Langfuse client using the credentials
 158        langfuse = Langfuse()
 159        ```
 160    """
 161
 162    log = logging.getLogger("langfuse")
 163    """Logger for the Langfuse client."""
 164
 165    host: str
 166    """Host of Langfuse API."""
 167
 168    project_id: Optional[str]
 169    """Project ID of the Langfuse project associated with the API keys provided."""
 170
 171    def __init__(
 172        self,
 173        public_key: Optional[str] = None,
 174        secret_key: Optional[str] = None,
 175        host: Optional[str] = None,
 176        release: Optional[str] = None,
 177        debug: bool = False,
 178        threads: Optional[int] = None,
 179        flush_at: Optional[int] = None,
 180        flush_interval: Optional[float] = None,
 181        max_retries: Optional[int] = None,
 182        timeout: Optional[int] = None,  # seconds
 183        sdk_integration: Optional[str] = "default",
 184        httpx_client: Optional[httpx.Client] = None,
 185        enabled: Optional[bool] = True,
 186        sample_rate: Optional[float] = None,
 187        mask: Optional[MaskFunction] = None,
 188    ):
 189        """Initialize the Langfuse client.
 190
 191        Args:
 192            public_key: Public API key of Langfuse project. Can be set via `LANGFUSE_PUBLIC_KEY` environment variable.
 193            secret_key: Secret API key of Langfuse project. Can be set via `LANGFUSE_SECRET_KEY` environment variable.
 194            host: Host of Langfuse API. Can be set via `LANGFUSE_HOST` environment variable. Defaults to `https://cloud.langfuse.com`.
 195            release: Release number/hash of the application to provide analytics grouped by release. Can be set via `LANGFUSE_RELEASE` environment variable.
 196            debug: Enables debug mode for more verbose logging. Can be set via `LANGFUSE_DEBUG` environment variable.
 197            threads: Number of consumer threads to execute network requests. Helps scaling the SDK for high load. Only increase this if you run into scaling issues.
 198            flush_at: Max batch size that's sent to the API.
 199            flush_interval: Max delay until a new batch is sent to the API.
 200            max_retries: Max number of retries in case of API/network errors.
 201            timeout: Timeout of API requests in seconds. Defaults to 20 seconds.
 202            httpx_client: Pass your own httpx client for more customizability of requests.
 203            sdk_integration: Used by intgerations that wrap the Langfuse SDK to add context for debugging and support. Not to be used directly.
 204            enabled: Enables or disables the Langfuse client. If disabled, all observability calls to the backend will be no-ops.
 205            sample_rate: Sampling rate for tracing. If set to 0.2, only 20% of the data will be sent to the backend. Can be set via `LANGFUSE_SAMPLE_RATE` environment variable.
 206            mask (langfuse.types.MaskFunction): Masking function for 'input' and 'output' fields in events. Function must take a single keyword argument `data` and return a serializable, masked version of the data.
 207
 208        Raises:
 209            ValueError: If public_key or secret_key are not set and not found in environment variables.
 210
 211        Example:
 212            Initiating the Langfuse client should always be first step to use Langfuse.
 213            ```python
 214            import os
 215            from langfuse import Langfuse
 216
 217            # Set the public and secret keys as environment variables
 218            os.environ['LANGFUSE_PUBLIC_KEY'] = public_key
 219            os.environ['LANGFUSE_SECRET_KEY'] = secret_key
 220
 221            # Initialize the Langfuse client using the credentials
 222            langfuse = Langfuse()
 223            ```
 224        """
 225        self.enabled = enabled
 226        public_key = public_key or os.environ.get("LANGFUSE_PUBLIC_KEY")
 227        secret_key = secret_key or os.environ.get("LANGFUSE_SECRET_KEY")
 228        sample_rate = (
 229            sample_rate
 230            if sample_rate
 231            is not None  # needs explicit None check, as 0 is a valid value
 232            else float(os.environ.get("LANGFUSE_SAMPLE_RATE", 1.0))
 233        )
 234
 235        if sample_rate is not None and (
 236            sample_rate > 1 or sample_rate < 0
 237        ):  # default value 1 will be set in the taskmanager
 238            self.enabled = False
 239            self.log.warning(
 240                "Langfuse client is disabled since the sample rate provided is not between 0 and 1."
 241            )
 242
 243        threads = threads or int(os.environ.get("LANGFUSE_THREADS", 1))
 244        flush_at = flush_at or int(os.environ.get("LANGFUSE_FLUSH_AT", 15))
 245        flush_interval = flush_interval or float(
 246            os.environ.get("LANGFUSE_FLUSH_INTERVAL", 0.5)
 247        )
 248
 249        max_retries = max_retries or int(os.environ.get("LANGFUSE_MAX_RETRIES", 3))
 250        timeout = timeout or int(os.environ.get("LANGFUSE_TIMEOUT", 20))
 251
 252        if not self.enabled:
 253            self.log.warning(
 254                "Langfuse client is disabled. No observability data will be sent."
 255            )
 256
 257        elif not public_key:
 258            self.enabled = False
 259            self.log.warning(
 260                "Langfuse client is disabled since no public_key was provided as a parameter or environment variable 'LANGFUSE_PUBLIC_KEY'. See our docs: https://langfuse.com/docs/sdk/python/low-level-sdk#initialize-client"
 261            )
 262
 263        elif not secret_key:
 264            self.enabled = False
 265            self.log.warning(
 266                "Langfuse client is disabled since no secret_key was provided as a parameter or environment variable 'LANGFUSE_SECRET_KEY'. See our docs: https://langfuse.com/docs/sdk/python/low-level-sdk#initialize-client"
 267            )
 268
 269        set_debug = debug if debug else (os.getenv("LANGFUSE_DEBUG", "False") == "True")
 270
 271        if set_debug is True:
 272            # Ensures that debug level messages are logged when debug mode is on.
 273            # Otherwise, defaults to WARNING level.
 274            # See https://docs.python.org/3/howto/logging.html#what-happens-if-no-configuration-is-provided
 275            logging.basicConfig()
 276            # Set level for all loggers under langfuse package
 277            logging.getLogger("langfuse").setLevel(logging.DEBUG)
 278
 279            clean_logger()
 280        else:
 281            logging.getLogger("langfuse").setLevel(logging.WARNING)
 282            clean_logger()
 283
 284        self.base_url = (
 285            host
 286            if host
 287            else os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com")
 288        )
 289
 290        self.httpx_client = httpx_client or httpx.Client(timeout=timeout)
 291
 292        public_api_client = FernLangfuse(
 293            base_url=self.base_url,
 294            username=public_key,
 295            password=secret_key,
 296            x_langfuse_sdk_name="python",
 297            x_langfuse_sdk_version=version,
 298            x_langfuse_public_key=public_key,
 299            httpx_client=self.httpx_client,
 300            timeout=timeout,
 301        )
 302        async_public_api_client = AsyncFernLangfuse(
 303            base_url=self.base_url,
 304            username=public_key,
 305            password=secret_key,
 306            x_langfuse_sdk_name="python",
 307            x_langfuse_sdk_version=version,
 308            x_langfuse_public_key=public_key,
 309            timeout=timeout,
 310        )
 311
 312        self.api = public_api_client
 313        self.client = public_api_client  # legacy, to be removed in next major release
 314        self.async_api = async_public_api_client
 315
 316        langfuse_client = LangfuseClient(
 317            public_key=public_key,
 318            secret_key=secret_key,
 319            base_url=self.base_url,
 320            version=version,
 321            timeout=timeout,
 322            session=self.httpx_client,
 323        )
 324
 325        args = {
 326            "threads": threads,
 327            "flush_at": flush_at,
 328            "flush_interval": flush_interval,
 329            "max_retries": max_retries,
 330            "client": langfuse_client,
 331            "api_client": self.client,
 332            "public_key": public_key,
 333            "sdk_name": "python",
 334            "sdk_version": version,
 335            "sdk_integration": sdk_integration,
 336            "enabled": self.enabled,
 337            "sample_rate": sample_rate,
 338            "mask": mask,
 339        }
 340
 341        self.task_manager = TaskManager(**args)
 342
 343        self.trace_id = None
 344        self.project_id = None
 345
 346        self.release = self._get_release_value(release)
 347
 348        self.prompt_cache = PromptCache()
 349
 350    def _get_release_value(self, release: Optional[str] = None) -> Optional[str]:
 351        if release:
 352            return release
 353        elif "LANGFUSE_RELEASE" in os.environ:
 354            return os.environ["LANGFUSE_RELEASE"]
 355        else:
 356            return get_common_release_envs()
 357
 358    def _get_project_id(self) -> Optional[str]:
 359        """Fetch and return the current project id. Persisted across requests. Returns None if no project id is found for api keys."""
 360        if not self.project_id:
 361            proj = self.client.projects.get()
 362            if not proj.data or not proj.data[0].id:
 363                return None
 364
 365            self.project_id = proj.data[0].id
 366
 367        return self.project_id
 368
 369    def get_trace_id(self) -> str:
 370        """Get the current trace id."""
 371        return self.trace_id
 372
 373    def get_trace_url(self) -> str:
 374        """Get the URL of the current trace to view it in the Langfuse UI."""
 375        project_id = self._get_project_id()
 376        if not project_id:
 377            return f"{self.base_url}/trace/{self.trace_id}"
 378
 379        return f"{self.base_url}/project/{project_id}/traces/{self.trace_id}"
 380
 381    def get_dataset(
 382        self, name: str, *, fetch_items_page_size: Optional[int] = 50
 383    ) -> "DatasetClient":
 384        """Fetch a dataset by its name.
 385
 386        Args:
 387            name (str): The name of the dataset to fetch.
 388            fetch_items_page_size (Optional[int]): All items of the dataset will be fetched in chunks of this size. Defaults to 50.
 389
 390        Returns:
 391            DatasetClient: The dataset with the given name.
 392        """
 393        try:
 394            self.log.debug(f"Getting datasets {name}")
 395            dataset = self.client.datasets.get(dataset_name=name)
 396
 397            dataset_items = []
 398            page = 1
 399            while True:
 400                new_items = self.client.dataset_items.list(
 401                    dataset_name=self._url_encode(name),
 402                    page=page,
 403                    limit=fetch_items_page_size,
 404                )
 405                dataset_items.extend(new_items.data)
 406                if new_items.meta.total_pages <= page:
 407                    break
 408                page += 1
 409
 410            items = [DatasetItemClient(i, langfuse=self) for i in dataset_items]
 411
 412            return DatasetClient(dataset, items=items)
 413        except Exception as e:
 414            handle_fern_exception(e)
 415            raise e
 416
 417    def get_dataset_item(self, id: str) -> "DatasetItemClient":
 418        """Get the dataset item with the given id."""
 419        try:
 420            self.log.debug(f"Getting dataset item {id}")
 421            dataset_item = self.client.dataset_items.get(id=id)
 422            return DatasetItemClient(dataset_item, langfuse=self)
 423        except Exception as e:
 424            handle_fern_exception(e)
 425            raise e
 426
 427    def auth_check(self) -> bool:
 428        """Check if the provided credentials (public and secret key) are valid.
 429
 430        Raises:
 431            Exception: If no projects were found for the provided credentials.
 432
 433        Note:
 434            This method is blocking. It is discouraged to use it in production code.
 435        """
 436        try:
 437            projects = self.client.projects.get()
 438            self.log.debug(
 439                f"Auth check successful, found {len(projects.data)} projects"
 440            )
 441            if len(projects.data) == 0:
 442                raise Exception(
 443                    "Auth check failed, no project found for the keys provided."
 444                )
 445            return True
 446
 447        except Exception as e:
 448            handle_fern_exception(e)
 449            raise e
 450
 451    def get_dataset_runs(
 452        self,
 453        dataset_name: str,
 454        *,
 455        page: typing.Optional[int] = None,
 456        limit: typing.Optional[int] = None,
 457    ) -> PaginatedDatasetRuns:
 458        """Get all dataset runs.
 459
 460        Args:
 461            dataset_name (str): Name of the dataset.
 462            page (Optional[int]): Page number of the dataset runs to return, starts at 1. Defaults to None.
 463            limit (Optional[int]): Maximum number of dataset runs to return. Defaults to 50.
 464
 465        Returns:
 466            PaginatedDatasetRuns: The dataset runs.
 467        """
 468        try:
 469            self.log.debug("Getting dataset runs")
 470            return self.client.datasets.get_runs(
 471                dataset_name=self._url_encode(dataset_name), page=page, limit=limit
 472            )
 473        except Exception as e:
 474            handle_fern_exception(e)
 475            raise e
 476
 477    def get_dataset_run(
 478        self,
 479        dataset_name: str,
 480        dataset_run_name: str,
 481    ) -> DatasetRunWithItems:
 482        """Get a dataset run.
 483
 484        Args:
 485            dataset_name: Name of the dataset.
 486            dataset_run_name: Name of the dataset run.
 487
 488        Returns:
 489            DatasetRunWithItems: The dataset run.
 490        """
 491        try:
 492            self.log.debug(
 493                f"Getting dataset runs for dataset {dataset_name} and run {dataset_run_name}"
 494            )
 495            return self.client.datasets.get_run(
 496                dataset_name=self._url_encode(dataset_name),
 497                run_name=self._url_encode(dataset_run_name),
 498            )
 499        except Exception as e:
 500            handle_fern_exception(e)
 501            raise e
 502
 503    def create_dataset(
 504        self,
 505        name: str,
 506        description: Optional[str] = None,
 507        metadata: Optional[Any] = None,
 508    ) -> Dataset:
 509        """Create a dataset with the given name on Langfuse.
 510
 511        Args:
 512            name: Name of the dataset to create.
 513            description: Description of the dataset. Defaults to None.
 514            metadata: Additional metadata. Defaults to None.
 515
 516        Returns:
 517            Dataset: The created dataset as returned by the Langfuse API.
 518        """
 519        try:
 520            body = CreateDatasetRequest(
 521                name=name, description=description, metadata=metadata
 522            )
 523            self.log.debug(f"Creating datasets {body}")
 524            return self.client.datasets.create(request=body)
 525        except Exception as e:
 526            handle_fern_exception(e)
 527            raise e
 528
 529    def create_dataset_item(
 530        self,
 531        dataset_name: str,
 532        input: Optional[Any] = None,
 533        expected_output: Optional[Any] = None,
 534        metadata: Optional[Any] = None,
 535        source_trace_id: Optional[str] = None,
 536        source_observation_id: Optional[str] = None,
 537        status: Optional[DatasetStatus] = None,
 538        id: Optional[str] = None,
 539    ) -> DatasetItem:
 540        """Create a dataset item.
 541
 542        Upserts if an item with id already exists.
 543
 544        Args:
 545            dataset_name: Name of the dataset in which the dataset item should be created.
 546            input: Input data. Defaults to None. Can contain any dict, list or scalar.
 547            expected_output: Expected output data. Defaults to None. Can contain any dict, list or scalar.
 548            metadata: Additional metadata. Defaults to None. Can contain any dict, list or scalar.
 549            source_trace_id: Id of the source trace. Defaults to None.
 550            source_observation_id: Id of the source observation. Defaults to None.
 551            status: Status of the dataset item. Defaults to ACTIVE for newly created items.
 552            id: Id of the dataset item. Defaults to None. Provide your own id if you want to dedupe dataset items. Id needs to be globally unique and cannot be reused across datasets.
 553
 554        Returns:
 555            DatasetItem: The created dataset item as returned by the Langfuse API.
 556
 557        Example:
 558            ```python
 559            from langfuse import Langfuse
 560
 561            langfuse = Langfuse()
 562
 563            # Uploading items to the Langfuse dataset named "capital_cities"
 564            langfuse.create_dataset_item(
 565                dataset_name="capital_cities",
 566                input={"input": {"country": "Italy"}},
 567                expected_output={"expected_output": "Rome"},
 568                metadata={"foo": "bar"}
 569            )
 570            ```
 571        """
 572        try:
 573            body = CreateDatasetItemRequest(
 574                datasetName=dataset_name,
 575                input=input,
 576                expectedOutput=expected_output,
 577                metadata=metadata,
 578                sourceTraceId=source_trace_id,
 579                sourceObservationId=source_observation_id,
 580                status=status,
 581                id=id,
 582            )
 583            self.log.debug(f"Creating dataset item {body}")
 584            return self.client.dataset_items.create(request=body)
 585        except Exception as e:
 586            handle_fern_exception(e)
 587            raise e
 588
 589    def fetch_trace(
 590        self,
 591        id: str,
 592    ) -> FetchTraceResponse:
 593        """Fetch a trace via the Langfuse API by its id.
 594
 595        Args:
 596            id: The id of the trace to fetch.
 597
 598        Returns:
 599            FetchTraceResponse: The trace with full details as returned by the Langfuse API on `data`.
 600
 601        Raises:
 602            Exception: If the trace with the given id could not be found within the authenticated project or if an error occurred during the request.
 603        """
 604        try:
 605            self.log.debug(f"Getting trace {id}")
 606            trace = self.client.trace.get(id)
 607            return FetchTraceResponse(data=trace)
 608        except Exception as e:
 609            handle_fern_exception(e)
 610            raise e
 611
 612    def get_trace(
 613        self,
 614        id: str,
 615    ) -> TraceWithFullDetails:
 616        """Get a trace via the Langfuse API by its id. Deprecated, use fetch_trace instead.
 617
 618        Args:
 619            id: The id of the trace to fetch.
 620
 621        Returns:
 622            TraceWithFullDetails: The trace with full details as returned by the Langfuse API.
 623
 624        Raises:
 625            Exception: If the trace with the given id could not be found within the authenticated project or if an error occurred during the request.
 626        """
 627        warnings.warn(
 628            "get_trace is deprecated, use fetch_trace instead.",
 629            DeprecationWarning,
 630        )
 631
 632        try:
 633            self.log.debug(f"Getting trace {id}")
 634            return self.client.trace.get(id)
 635        except Exception as e:
 636            handle_fern_exception(e)
 637            raise e
 638
 639    def fetch_traces(
 640        self,
 641        *,
 642        page: Optional[int] = None,
 643        limit: Optional[int] = None,
 644        user_id: Optional[str] = None,
 645        name: Optional[str] = None,
 646        session_id: Optional[str] = None,
 647        from_timestamp: Optional[dt.datetime] = None,
 648        to_timestamp: Optional[dt.datetime] = None,
 649        order_by: Optional[str] = None,
 650        tags: Optional[Union[str, Sequence[str]]] = None,
 651    ) -> FetchTracesResponse:
 652        """Fetch a list of traces in the current project matching the given parameters.
 653
 654        Args:
 655            page (Optional[int]): Page number, starts at 1. Defaults to None.
 656            limit (Optional[int]): Limit of items per page. If you encounter API issues due to too large page sizes, try to reduce the limit. Defaults to None.
 657            name (Optional[str]): Filter by name of traces. Defaults to None.
 658            user_id (Optional[str]): Filter by user_id. Defaults to None.
 659            session_id (Optional[str]): Filter by session_id. Defaults to None.
 660            from_timestamp (Optional[dt.datetime]): Retrieve only traces with a timestamp on or after this datetime. Defaults to None.
 661            to_timestamp (Optional[dt.datetime]): Retrieve only traces with a timestamp before this datetime. Defaults to None.
 662            order_by (Optional[str]): Format of the string `[field].[asc/desc]`. Fields: id, timestamp, name, userId, release, version, public, bookmarked, sessionId. Example: `timestamp.asc`. Defaults to None.
 663            tags (Optional[Union[str, Sequence[str]]]): Filter by tags. Defaults to None.
 664
 665        Returns:
 666            FetchTracesResponse, list of traces on `data` and metadata on `meta`.
 667
 668        Raises:
 669            Exception: If an error occurred during the request.
 670        """
 671        try:
 672            self.log.debug(
 673                f"Getting traces... {page}, {limit}, {name}, {user_id}, {session_id}, {from_timestamp}, {to_timestamp}, {order_by}, {tags}"
 674            )
 675            res = self.client.trace.list(
 676                page=page,
 677                limit=limit,
 678                name=name,
 679                user_id=user_id,
 680                session_id=session_id,
 681                from_timestamp=from_timestamp,
 682                to_timestamp=to_timestamp,
 683                order_by=order_by,
 684                tags=tags,
 685            )
 686            return FetchTracesResponse(data=res.data, meta=res.meta)
 687        except Exception as e:
 688            handle_fern_exception(e)
 689            raise e
 690
 691    def get_traces(
 692        self,
 693        *,
 694        page: Optional[int] = None,
 695        limit: Optional[int] = None,
 696        user_id: Optional[str] = None,
 697        name: Optional[str] = None,
 698        session_id: Optional[str] = None,
 699        from_timestamp: Optional[dt.datetime] = None,
 700        to_timestamp: Optional[dt.datetime] = None,
 701        order_by: Optional[str] = None,
 702        tags: Optional[Union[str, Sequence[str]]] = None,
 703    ) -> Traces:
 704        """Get a list of traces in the current project matching the given parameters. Deprecated, use fetch_traces instead.
 705
 706        Args:
 707            page (Optional[int]): Page number, starts at 1. Defaults to None.
 708            limit (Optional[int]): Limit of items per page. If you encounter API issues due to too large page sizes, try to reduce the limit. Defaults to None.
 709            name (Optional[str]): Filter by name of traces. Defaults to None.
 710            user_id (Optional[str]): Filter by user_id. Defaults to None.
 711            session_id (Optional[str]): Filter by session_id. Defaults to None.
 712            from_timestamp (Optional[dt.datetime]): Retrieve only traces with a timestamp on or after this datetime. Defaults to None.
 713            to_timestamp (Optional[dt.datetime]): Retrieve only traces with a timestamp before this datetime. Defaults to None.
 714            order_by (Optional[str]): Format of the string `[field].[asc/desc]`. Fields: id, timestamp, name, userId, release, version, public, bookmarked, sessionId. Example: `timestamp.asc`. Defaults to None.
 715            tags (Optional[Union[str, Sequence[str]]]): Filter by tags. Defaults to None.
 716
 717        Returns:
 718            List of Traces
 719
 720        Raises:
 721            Exception: If an error occurred during the request.
 722        """
 723        warnings.warn(
 724            "get_traces is deprecated, use fetch_traces instead.",
 725            DeprecationWarning,
 726        )
 727        try:
 728            self.log.debug(
 729                f"Getting traces... {page}, {limit}, {name}, {user_id}, {session_id}, {from_timestamp}, {to_timestamp}, {order_by}, {tags}"
 730            )
 731            return self.client.trace.list(
 732                page=page,
 733                limit=limit,
 734                name=name,
 735                user_id=user_id,
 736                session_id=session_id,
 737                from_timestamp=from_timestamp,
 738                to_timestamp=to_timestamp,
 739                order_by=order_by,
 740                tags=tags,
 741            )
 742        except Exception as e:
 743            handle_fern_exception(e)
 744            raise e
 745
 746    def fetch_observations(
 747        self,
 748        *,
 749        page: typing.Optional[int] = None,
 750        limit: typing.Optional[int] = None,
 751        name: typing.Optional[str] = None,
 752        user_id: typing.Optional[str] = None,
 753        trace_id: typing.Optional[str] = None,
 754        parent_observation_id: typing.Optional[str] = None,
 755        from_start_time: typing.Optional[dt.datetime] = None,
 756        to_start_time: typing.Optional[dt.datetime] = None,
 757        type: typing.Optional[str] = None,
 758    ) -> FetchObservationsResponse:
 759        """Get a list of observations in the current project matching the given parameters.
 760
 761        Args:
 762            page (Optional[int]): Page number of the observations to return. Defaults to None.
 763            limit (Optional[int]): Maximum number of observations to return. Defaults to None.
 764            name (Optional[str]): Name of the observations to return. Defaults to None.
 765            user_id (Optional[str]): User identifier. Defaults to None.
 766            trace_id (Optional[str]): Trace identifier. Defaults to None.
 767            parent_observation_id (Optional[str]): Parent observation identifier. Defaults to None.
 768            from_start_time (Optional[dt.datetime]): Retrieve only observations with a start_time on or after this datetime. Defaults to None.
 769            to_start_time (Optional[dt.datetime]): Retrieve only observations with a start_time before this datetime. Defaults to None.
 770            type (Optional[str]): Type of the observation. Defaults to None.
 771
 772        Returns:
 773            FetchObservationsResponse, list of observations on `data` and metadata on `meta`.
 774
 775        Raises:
 776            Exception: If an error occurred during the request.
 777        """
 778        try:
 779            self.log.debug(
 780                f"Getting observations... {page}, {limit}, {name}, {user_id}, {trace_id}, {parent_observation_id}, {from_start_time}, {to_start_time}, {type}"
 781            )
 782            res = self.client.observations.get_many(
 783                page=page,
 784                limit=limit,
 785                name=name,
 786                user_id=user_id,
 787                trace_id=trace_id,
 788                parent_observation_id=parent_observation_id,
 789                from_start_time=from_start_time,
 790                to_start_time=to_start_time,
 791                type=type,
 792            )
 793            return FetchObservationsResponse(data=res.data, meta=res.meta)
 794        except Exception as e:
 795            self.log.exception(e)
 796            raise e
 797
 798    def get_observations(
 799        self,
 800        *,
 801        page: typing.Optional[int] = None,
 802        limit: typing.Optional[int] = None,
 803        name: typing.Optional[str] = None,
 804        user_id: typing.Optional[str] = None,
 805        trace_id: typing.Optional[str] = None,
 806        parent_observation_id: typing.Optional[str] = None,
 807        from_start_time: typing.Optional[dt.datetime] = None,
 808        to_start_time: typing.Optional[dt.datetime] = None,
 809        type: typing.Optional[str] = None,
 810    ) -> ObservationsViews:
 811        """Get a list of observations in the current project matching the given parameters. Deprecated, use fetch_observations instead.
 812
 813        Args:
 814            page (Optional[int]): Page number of the observations to return. Defaults to None.
 815            limit (Optional[int]): Maximum number of observations to return. Defaults to None.
 816            name (Optional[str]): Name of the observations to return. Defaults to None.
 817            user_id (Optional[str]): User identifier. Defaults to None.
 818            trace_id (Optional[str]): Trace identifier. Defaults to None.
 819            parent_observation_id (Optional[str]): Parent observation identifier. Defaults to None.
 820            from_start_time (Optional[dt.datetime]): Retrieve only observations with a start_time on or after this datetime. Defaults to None.
 821            to_start_time (Optional[dt.datetime]): Retrieve only observations with a start_time before this datetime. Defaults to None.
 822            type (Optional[str]): Type of the observation. Defaults to None.
 823
 824        Returns:
 825            List of ObservationsViews: List of observations in the project matching the given parameters.
 826
 827        Raises:
 828            Exception: If an error occurred during the request.
 829        """
 830        warnings.warn(
 831            "get_observations is deprecated, use fetch_observations instead.",
 832            DeprecationWarning,
 833        )
 834        try:
 835            self.log.debug(
 836                f"Getting observations... {page}, {limit}, {name}, {user_id}, {trace_id}, {parent_observation_id}, {from_start_time}, {to_start_time}, {type}"
 837            )
 838            return self.client.observations.get_many(
 839                page=page,
 840                limit=limit,
 841                name=name,
 842                user_id=user_id,
 843                trace_id=trace_id,
 844                parent_observation_id=parent_observation_id,
 845                from_start_time=from_start_time,
 846                to_start_time=to_start_time,
 847                type=type,
 848            )
 849        except Exception as e:
 850            handle_fern_exception(e)
 851            raise e
 852
 853    def get_generations(
 854        self,
 855        *,
 856        page: typing.Optional[int] = None,
 857        limit: typing.Optional[int] = None,
 858        name: typing.Optional[str] = None,
 859        user_id: typing.Optional[str] = None,
 860        trace_id: typing.Optional[str] = None,
 861        from_start_time: typing.Optional[dt.datetime] = None,
 862        to_start_time: typing.Optional[dt.datetime] = None,
 863        parent_observation_id: typing.Optional[str] = None,
 864    ) -> ObservationsViews:
 865        """Get a list of generations in the current project matching the given parameters. Deprecated, use fetch_observations(type='GENERATION') instead.
 866
 867        Args:
 868            page (Optional[int]): Page number of the generations to return. Defaults to None.
 869            limit (Optional[int]): Maximum number of generations to return. Defaults to None.
 870            name (Optional[str]): Name of the generations to return. Defaults to None.
 871            user_id (Optional[str]): User identifier of the generations to return. Defaults to None.
 872            trace_id (Optional[str]): Trace identifier of the generations to return. Defaults to None.
 873            from_start_time (Optional[dt.datetime]): Retrieve only observations with a start_time on or after this datetime. Defaults to None.
 874            to_start_time (Optional[dt.datetime]): Retrieve only observations with a start_time before this datetime. Defaults to None.
 875            parent_observation_id (Optional[str]): Parent observation identifier of the generations to return. Defaults to None.
 876
 877        Returns:
 878            List of ObservationsViews: List of generations in the project matching the given parameters.
 879
 880        Raises:
 881            Exception: If an error occurred during the request.
 882        """
 883        warnings.warn(
 884            "get_generations is deprecated, use `fetch_observations(type='GENERATION')` instead.",
 885            DeprecationWarning,
 886        )
 887        return self.get_observations(
 888            page=page,
 889            limit=limit,
 890            name=name,
 891            user_id=user_id,
 892            trace_id=trace_id,
 893            parent_observation_id=parent_observation_id,
 894            from_start_time=from_start_time,
 895            to_start_time=to_start_time,
 896            type="GENERATION",
 897        )
 898
 899    def fetch_observation(
 900        self,
 901        id: str,
 902    ) -> FetchObservationResponse:
 903        """Get an observation in the current project with the given identifier.
 904
 905        Args:
 906            id: The identifier of the observation to fetch.
 907
 908        Returns:
 909            FetchObservationResponse: The observation with the given id on `data`.
 910
 911        Raises:
 912            Exception: If the observation with the given id could not be found within the authenticated project or if an error occurred during the request.
 913        """
 914        try:
 915            self.log.debug(f"Getting observation {id}")
 916            observation = self.client.observations.get(id)
 917            return FetchObservationResponse(data=observation)
 918        except Exception as e:
 919            handle_fern_exception(e)
 920            raise e
 921
 922    def fetch_media(self, id: str) -> FetchMediaResponse:
 923        """Get media content by ID.
 924
 925        Args:
 926            id: The identifier of the media content to fetch.
 927
 928        Returns:
 929            FetchMediaResponse: The media data of the given id on `data`.
 930
 931        Raises:
 932            Exception: If the media content with the given id could not be found within the authenticated project or if an error occurred during the request.
 933        """
 934        try:
 935            return FetchMediaResponse(data=self.client.media.get(id))
 936        except Exception as e:
 937            handle_fern_exception(e)
 938            raise e
 939
 940    def resolve_media_references(
 941        self,
 942        *,
 943        obj: Any,
 944        resolve_with: Literal["base64_data_uri"],
 945        max_depth: int = 10,
 946        content_fetch_timeout_seconds: int = 10,
 947    ):
 948        """Replace media reference strings in an object with base64 data URIs.
 949
 950        This method recursively traverses an object (up to max_depth) looking for media reference strings
 951        in the format "@@@langfuseMedia:...@@@". When found, it (synchronously) fetches the actual media content using
 952        the provided Langfuse client and replaces the reference string with a base64 data URI.
 953
 954        If fetching media content fails for a reference string, a warning is logged and the reference
 955        string is left unchanged.
 956
 957        Args:
 958            obj: The object to process. Can be a primitive value, array, or nested object.
 959                If the object has a __dict__ attribute, a dict will be returned instead of the original object type.
 960            resolve_with: The representation of the media content to replace the media reference string with.
 961                Currently only "base64_data_uri" is supported.
 962            max_depth: int: The maximum depth to traverse the object. Default is 10.
 963            content_fetch_timeout_seconds: int: The timeout in seconds for fetching media content. Default is 10.
 964
 965        Returns:
 966            A deep copy of the input object with all media references replaced with base64 data URIs where possible.
 967            If the input object has a __dict__ attribute, a dict will be returned instead of the original object type.
 968
 969        Example:
 970            obj = {
 971                "image": "@@@langfuseMedia:type=image/jpeg|id=123|source=bytes@@@",
 972                "nested": {
 973                    "pdf": "@@@langfuseMedia:type=application/pdf|id=456|source=bytes@@@"
 974                }
 975            }
 976
 977            result = await LangfuseMedia.resolve_media_references(obj, langfuse_client)
 978
 979            # Result:
 980            # {
 981            #     "image": "data:image/jpeg;base64,/9j/4AAQSkZJRg...",
 982            #     "nested": {
 983            #         "pdf": "data:application/pdf;base64,JVBERi0xLjcK..."
 984            #     }
 985            # }
 986        """
 987        return LangfuseMedia.resolve_media_references(
 988            langfuse_client=self,
 989            obj=obj,
 990            resolve_with=resolve_with,
 991            max_depth=max_depth,
 992            content_fetch_timeout_seconds=content_fetch_timeout_seconds,
 993        )
 994
 995    def get_observation(
 996        self,
 997        id: str,
 998    ) -> Observation:
 999        """Get an observation in the current project with the given identifier. Deprecated, use fetch_observation instead.
1000
1001        Args:
1002            id: The identifier of the observation to fetch.
1003
1004        Raises:
1005            Exception: If the observation with the given id could not be found within the authenticated project or if an error occurred during the request.
1006        """
1007        warnings.warn(
1008            "get_observation is deprecated, use fetch_observation instead.",
1009            DeprecationWarning,
1010        )
1011        try:
1012            self.log.debug(f"Getting observation {id}")
1013            return self.client.observations.get(id)
1014        except Exception as e:
1015            handle_fern_exception(e)
1016            raise e
1017
1018    def fetch_sessions(
1019        self,
1020        *,
1021        page: typing.Optional[int] = None,
1022        limit: typing.Optional[int] = None,
1023        from_timestamp: typing.Optional[dt.datetime] = None,
1024        to_timestamp: typing.Optional[dt.datetime] = None,
1025    ) -> FetchSessionsResponse:
1026        """Get a list of sessions in the current project.
1027
1028        Args:
1029            page (Optional[int]): Page number of the sessions to return. Defaults to None.
1030            limit (Optional[int]): Maximum number of sessions to return. Defaults to None.
1031            from_timestamp (Optional[dt.datetime]): Retrieve only sessions with a timestamp on or after this datetime. Defaults to None.
1032            to_timestamp (Optional[dt.datetime]): Retrieve only sessions with a timestamp before this datetime. Defaults to None.
1033
1034        Returns:
1035            FetchSessionsResponse, list of sessions on `data` and metadata on `meta`.
1036
1037        Raises:
1038            Exception: If an error occurred during the request.
1039        """
1040        try:
1041            self.log.debug(
1042                f"Getting sessions... {page}, {limit}, {from_timestamp}, {to_timestamp}"
1043            )
1044            res = self.client.sessions.list(
1045                page=page,
1046                limit=limit,
1047                from_timestamp=from_timestamp,
1048                to_timestamp=to_timestamp,
1049            )
1050            return FetchSessionsResponse(data=res.data, meta=res.meta)
1051        except Exception as e:
1052            handle_fern_exception(e)
1053            raise e
1054
1055    @overload
1056    def get_prompt(
1057        self,
1058        name: str,
1059        version: Optional[int] = None,
1060        *,
1061        label: Optional[str] = None,
1062        type: Literal["chat"],
1063        cache_ttl_seconds: Optional[int] = None,
1064        fallback: Optional[List[ChatMessageDict]] = None,
1065        max_retries: Optional[int] = None,
1066        fetch_timeout_seconds: Optional[int] = None,
1067    ) -> ChatPromptClient: ...
1068
1069    @overload
1070    def get_prompt(
1071        self,
1072        name: str,
1073        version: Optional[int] = None,
1074        *,
1075        label: Optional[str] = None,
1076        type: Literal["text"] = "text",
1077        cache_ttl_seconds: Optional[int] = None,
1078        fallback: Optional[str] = None,
1079        max_retries: Optional[int] = None,
1080        fetch_timeout_seconds: Optional[int] = None,
1081    ) -> TextPromptClient: ...
1082
1083    def get_prompt(
1084        self,
1085        name: str,
1086        version: Optional[int] = None,
1087        *,
1088        label: Optional[str] = None,
1089        type: Literal["chat", "text"] = "text",
1090        cache_ttl_seconds: Optional[int] = None,
1091        fallback: Union[Optional[List[ChatMessageDict]], Optional[str]] = None,
1092        max_retries: Optional[int] = None,
1093        fetch_timeout_seconds: Optional[int] = None,
1094    ) -> PromptClient:
1095        """Get a prompt.
1096
1097        This method attempts to fetch the requested prompt from the local cache. If the prompt is not found
1098        in the cache or if the cached prompt has expired, it will try to fetch the prompt from the server again
1099        and update the cache. If fetching the new prompt fails, and there is an expired prompt in the cache, it will
1100        return the expired prompt as a fallback.
1101
1102        Args:
1103            name (str): The name of the prompt to retrieve.
1104
1105        Keyword Args:
1106            version (Optional[int]): The version of the prompt to retrieve. If no label and version is specified, the `production` label is returned. Specify either version or label, not both.
1107            label: Optional[str]: The label of the prompt to retrieve. If no label and version is specified, the `production` label is returned. Specify either version or label, not both.
1108            cache_ttl_seconds: Optional[int]: Time-to-live in seconds for caching the prompt. Must be specified as a
1109            keyword argument. If not set, defaults to 60 seconds. Disables caching if set to 0.
1110            type: Literal["chat", "text"]: The type of the prompt to retrieve. Defaults to "text".
1111            fallback: Union[Optional[List[ChatMessageDict]], Optional[str]]: The prompt string to return if fetching the prompt fails. Important on the first call where no cached prompt is available. Follows Langfuse prompt formatting with double curly braces for variables. Defaults to None.
1112            max_retries: Optional[int]: The maximum number of retries in case of API/network errors. Defaults to 2. The maximum value is 4. Retries have an exponential backoff with a maximum delay of 10 seconds.
1113            fetch_timeout_seconds: Optional[int]: The timeout in milliseconds for fetching the prompt. Defaults to the default timeout set on the SDK, which is 10 seconds per default.
1114
1115        Returns:
1116            The prompt object retrieved from the cache or directly fetched if not cached or expired of type
1117            - TextPromptClient, if type argument is 'text'.
1118            - ChatPromptClient, if type argument is 'chat'.
1119
1120        Raises:
1121            Exception: Propagates any exceptions raised during the fetching of a new prompt, unless there is an
1122            expired prompt in the cache, in which case it logs a warning and returns the expired prompt.
1123        """
1124        if version is not None and label is not None:
1125            raise ValueError("Cannot specify both version and label at the same time.")
1126
1127        if not name:
1128            raise ValueError("Prompt name cannot be empty.")
1129
1130        cache_key = PromptCache.generate_cache_key(name, version=version, label=label)
1131        bounded_max_retries = self._get_bounded_max_retries(
1132            max_retries, default_max_retries=2, max_retries_upper_bound=4
1133        )
1134
1135        self.log.debug(f"Getting prompt '{cache_key}'")
1136        cached_prompt = self.prompt_cache.get(cache_key)
1137
1138        if cached_prompt is None or cache_ttl_seconds == 0:
1139            self.log.debug(
1140                f"Prompt '{cache_key}' not found in cache or caching disabled."
1141            )
1142            try:
1143                return self._fetch_prompt_and_update_cache(
1144                    name,
1145                    version=version,
1146                    label=label,
1147                    ttl_seconds=cache_ttl_seconds,
1148                    max_retries=bounded_max_retries,
1149                    fetch_timeout_seconds=fetch_timeout_seconds,
1150                )
1151            except Exception as e:
1152                if fallback:
1153                    self.log.warning(
1154                        f"Returning fallback prompt for '{cache_key}' due to fetch error: {e}"
1155                    )
1156
1157                    fallback_client_args = {
1158                        "name": name,
1159                        "prompt": fallback,
1160                        "type": type,
1161                        "version": version or 0,
1162                        "config": {},
1163                        "labels": [label] if label else [],
1164                        "tags": [],
1165                    }
1166
1167                    if type == "text":
1168                        return TextPromptClient(
1169                            prompt=Prompt_Text(**fallback_client_args),
1170                            is_fallback=True,
1171                        )
1172
1173                    if type == "chat":
1174                        return ChatPromptClient(
1175                            prompt=Prompt_Chat(**fallback_client_args),
1176                            is_fallback=True,
1177                        )
1178
1179                raise e
1180
1181        if cached_prompt.is_expired():
1182            self.log.debug(f"Stale prompt '{cache_key}' found in cache.")
1183            try:
1184                # refresh prompt in background thread, refresh_prompt deduplicates tasks
1185                self.log.debug(f"Refreshing prompt '{cache_key}' in background.")
1186                self.prompt_cache.add_refresh_prompt_task(
1187                    cache_key,
1188                    lambda: self._fetch_prompt_and_update_cache(
1189                        name,
1190                        version=version,
1191                        label=label,
1192                        ttl_seconds=cache_ttl_seconds,
1193                        max_retries=bounded_max_retries,
1194                        fetch_timeout_seconds=fetch_timeout_seconds,
1195                    ),
1196                )
1197                self.log.debug(f"Returning stale prompt '{cache_key}' from cache.")
1198                # return stale prompt
1199                return cached_prompt.value
1200
1201            except Exception as e:
1202                self.log.warning(
1203                    f"Error when refreshing cached prompt '{cache_key}', returning cached version. Error: {e}"
1204                )
1205                # creation of refresh prompt task failed, return stale prompt
1206                return cached_prompt.value
1207
1208        return cached_prompt.value
1209
1210    def _fetch_prompt_and_update_cache(
1211        self,
1212        name: str,
1213        *,
1214        version: Optional[int] = None,
1215        label: Optional[str] = None,
1216        ttl_seconds: Optional[int] = None,
1217        max_retries: int,
1218        fetch_timeout_seconds,
1219    ) -> PromptClient:
1220        try:
1221            cache_key = PromptCache.generate_cache_key(
1222                name, version=version, label=label
1223            )
1224
1225            self.log.debug(f"Fetching prompt '{cache_key}' from server...")
1226
1227            @backoff.on_exception(
1228                backoff.constant, Exception, max_tries=max_retries, logger=None
1229            )
1230            def fetch_prompts():
1231                return self.client.prompts.get(
1232                    self._url_encode(name),
1233                    version=version,
1234                    label=label,
1235                    request_options={
1236                        "timeout_in_seconds": fetch_timeout_seconds,
1237                    }
1238                    if fetch_timeout_seconds is not None
1239                    else None,
1240                )
1241
1242            prompt_response = fetch_prompts()
1243
1244            if prompt_response.type == "chat":
1245                prompt = ChatPromptClient(prompt_response)
1246            else:
1247                prompt = TextPromptClient(prompt_response)
1248
1249            self.prompt_cache.set(cache_key, prompt, ttl_seconds)
1250
1251            return prompt
1252
1253        except Exception as e:
1254            self.log.error(f"Error while fetching prompt '{cache_key}': {str(e)}")
1255            raise e
1256
1257    def _get_bounded_max_retries(
1258        self,
1259        max_retries: Optional[int],
1260        *,
1261        default_max_retries: int = 2,
1262        max_retries_upper_bound: int = 4,
1263    ) -> int:
1264        if max_retries is None:
1265            return default_max_retries
1266
1267        bounded_max_retries = min(
1268            max(max_retries, 0),
1269            max_retries_upper_bound,
1270        )
1271
1272        return bounded_max_retries
1273
1274    @overload
1275    def create_prompt(
1276        self,
1277        *,
1278        name: str,
1279        prompt: List[ChatMessageDict],
1280        is_active: Optional[bool] = None,  # deprecated
1281        labels: List[str] = [],
1282        tags: Optional[List[str]] = None,
1283        type: Optional[Literal["chat"]],
1284        config: Optional[Any] = None,
1285        commit_message: Optional[str] = None,
1286    ) -> ChatPromptClient: ...
1287
1288    @overload
1289    def create_prompt(
1290        self,
1291        *,
1292        name: str,
1293        prompt: str,
1294        is_active: Optional[bool] = None,  # deprecated
1295        labels: List[str] = [],
1296        tags: Optional[List[str]] = None,
1297        type: Optional[Literal["text"]] = "text",
1298        config: Optional[Any] = None,
1299        commit_message: Optional[str] = None,
1300    ) -> TextPromptClient: ...
1301
1302    def create_prompt(
1303        self,
1304        *,
1305        name: str,
1306        prompt: Union[str, List[ChatMessageDict]],
1307        is_active: Optional[bool] = None,  # deprecated
1308        labels: List[str] = [],
1309        tags: Optional[List[str]] = None,
1310        type: Optional[Literal["chat", "text"]] = "text",
1311        config: Optional[Any] = None,
1312        commit_message: Optional[str] = None,
1313    ) -> PromptClient:
1314        """Create a new prompt in Langfuse.
1315
1316        Keyword Args:
1317            name : The name of the prompt to be created.
1318            prompt : The content of the prompt to be created.
1319            is_active [DEPRECATED] : A flag indicating whether the prompt is active or not. This is deprecated and will be removed in a future release. Please use the 'production' label instead.
1320            labels: The labels of the prompt. Defaults to None. To create a default-served prompt, add the 'production' label.
1321            tags: The tags of the prompt. Defaults to None. Will be applied to all versions of the prompt.
1322            config: Additional structured data to be saved with the prompt. Defaults to None.
1323            type: The type of the prompt to be created. "chat" vs. "text". Defaults to "text".
1324            commit_message: Optional string describing the change.
1325
1326        Returns:
1327            TextPromptClient: The prompt if type argument is 'text'.
1328            ChatPromptClient: The prompt if type argument is 'chat'.
1329        """
1330        try:
1331            self.log.debug(f"Creating prompt {name=}, {version=}, {labels=}")
1332
1333            # Handle deprecated is_active flag
1334            if is_active:
1335                self.log.warning(
1336                    "The 'is_active' flag is deprecated and will be removed in a future release. Please use the 'production' label instead."
1337                )
1338
1339                labels = labels if "production" in labels else labels + ["production"]
1340
1341            if type == "chat":
1342                if not isinstance(prompt, list):
1343                    raise ValueError(
1344                        "For 'chat' type, 'prompt' must be a list of chat messages with role and content attributes."
1345                    )
1346                request = CreatePromptRequest_Chat(
1347                    name=name,
1348                    prompt=prompt,
1349                    labels=labels,
1350                    tags=tags,
1351                    config=config or {},
1352                    commitMessage=commit_message,
1353                    type="chat",
1354                )
1355                server_prompt = self.client.prompts.create(request=request)
1356
1357                return ChatPromptClient(prompt=server_prompt)
1358
1359            if not isinstance(prompt, str):
1360                raise ValueError("For 'text' type, 'prompt' must be a string.")
1361
1362            request = CreatePromptRequest_Text(
1363                name=name,
1364                prompt=prompt,
1365                labels=labels,
1366                tags=tags,
1367                config=config or {},
1368                commitMessage=commit_message,
1369                type="text",
1370            )
1371
1372            server_prompt = self.client.prompts.create(request=request)
1373            return TextPromptClient(prompt=server_prompt)
1374
1375        except Exception as e:
1376            handle_fern_exception(e)
1377            raise e
1378
1379    def update_prompt(
1380        self,
1381        *,
1382        name: str,
1383        version: int,
1384        new_labels: List[str] = [],
1385    ):
1386        """Update an existing prompt version in Langfuse. The Langfuse SDK prompt cache is invalidated for all prompts witht he specified name.
1387
1388        Args:
1389            name (str): The name of the prompt to update.
1390            version (int): The version number of the prompt to update.
1391            new_labels (List[str], optional): New labels to assign to the prompt version. Labels are unique across versions. The "latest" label is reserved and managed by Langfuse. Defaults to [].
1392
1393        Returns:
1394            Prompt: The updated prompt from the Langfuse API.
1395
1396        """
1397        updated_prompt = self.client.prompt_version.update(
1398            name=name,
1399            version=version,
1400            new_labels=new_labels,
1401        )
1402        self.prompt_cache.invalidate(name)
1403        return updated_prompt
1404
1405    def _url_encode(self, url: str) -> str:
1406        return urllib.parse.quote(url)
1407
1408    def trace(
1409        self,
1410        *,
1411        id: typing.Optional[str] = None,
1412        name: typing.Optional[str] = None,
1413        user_id: typing.Optional[str] = None,
1414        session_id: typing.Optional[str] = None,
1415        version: typing.Optional[str] = None,
1416        input: typing.Optional[typing.Any] = None,
1417        output: typing.Optional[typing.Any] = None,
1418        metadata: typing.Optional[typing.Any] = None,
1419        tags: typing.Optional[typing.List[str]] = None,
1420        timestamp: typing.Optional[dt.datetime] = None,
1421        public: typing.Optional[bool] = None,
1422        **kwargs,
1423    ) -> "StatefulTraceClient":
1424        """Create a trace.
1425
1426        Args:
1427            id: The id of the trace can be set, defaults to a random id. Set it to link traces to external systems or when creating a distributed trace. Traces are upserted on id.
1428            name: Identifier of the trace. Useful for sorting/filtering in the UI.
1429            input: The input of the trace. Can be any JSON object.
1430            output: The output of the trace. Can be any JSON object.
1431            metadata: Additional metadata of the trace. Can be any JSON object. Metadata is merged when being updated via the API.
1432            user_id: The id of the user that triggered the execution. Used to provide user-level analytics.
1433            session_id: Used to group multiple traces into a session in Langfuse. Use your own session/thread identifier.
1434            version: The version of the trace type. Used to understand how changes to the trace type affect metrics. Useful in debugging.
1435            release: The release identifier of the current deployment. Used to understand how changes of different deployments affect metrics. Useful in debugging.
1436            tags: Tags are used to categorize or label traces. Traces can be filtered by tags in the UI and GET API. Tags can also be changed in the UI. Tags are merged and never deleted via the API.
1437            timestamp: The timestamp of the trace. Defaults to the current time if not provided.
1438            public: You can make a trace `public` to share it via a public link. This allows others to view the trace without needing to log in or be members of your Langfuse project.
1439            **kwargs: Additional keyword arguments that can be included in the trace.
1440
1441        Returns:
1442            StatefulTraceClient: The created trace.
1443
1444        Example:
1445            ```python
1446            from langfuse import Langfuse
1447
1448            langfuse = Langfuse()
1449
1450            trace = langfuse.trace(
1451                name="example-application",
1452                user_id="user-1234")
1453            )
1454            ```
1455        """
1456        new_id = id or str(uuid.uuid4())
1457        self.trace_id = new_id
1458        try:
1459            new_dict = {
1460                "id": new_id,
1461                "name": name,
1462                "userId": user_id,
1463                "sessionId": session_id
1464                or kwargs.get("sessionId", None),  # backward compatibility
1465                "release": self.release,
1466                "version": version,
1467                "metadata": metadata,
1468                "input": input,
1469                "output": output,
1470                "tags": tags,
1471                "timestamp": timestamp or _get_timestamp(),
1472                "public": public,
1473            }
1474            if kwargs is not None:
1475                new_dict.update(kwargs)
1476
1477            new_body = TraceBody(**new_dict)
1478
1479            self.log.debug(f"Creating trace {_filter_io_from_event_body(new_dict)}")
1480            event = {
1481                "id": str(uuid.uuid4()),
1482                "type": "trace-create",
1483                "body": new_body,
1484            }
1485
1486            self.task_manager.add_task(
1487                event,
1488            )
1489
1490        except Exception as e:
1491            self.log.exception(e)
1492        finally:
1493            self._log_memory_usage()
1494
1495            return StatefulTraceClient(
1496                self.client, new_id, StateType.TRACE, new_id, self.task_manager
1497            )
1498
1499    def _log_memory_usage(self):
1500        try:
1501            is_malloc_tracing_enabled = bool(int(os.getenv("PYTHONTRACEMALLOC", 0)))
1502            report_interval = int(os.getenv("LANGFUSE_DEBUG_MEMORY_REPORT_INTERVAL", 0))
1503            top_k_items = int(os.getenv("LANGFUSE_DEBUG_MEMORY_TOP_K", 10))
1504
1505            if (
1506                not is_malloc_tracing_enabled
1507                or report_interval <= 0
1508                or round(time.monotonic()) % report_interval != 0
1509            ):
1510                return
1511
1512            snapshot = tracemalloc.take_snapshot().statistics("lineno")
1513
1514            total_memory_usage = sum([stat.size for stat in snapshot]) / 1024 / 1024
1515            memory_usage_total_items = [f"{stat}" for stat in snapshot]
1516            memory_usage_langfuse_items = [
1517                stat for stat in memory_usage_total_items if "/langfuse/" in stat
1518            ]
1519
1520            logged_memory_usage = {
1521                "all_files": [f"{stat}" for stat in memory_usage_total_items][
1522                    :top_k_items
1523                ],
1524                "langfuse_files": [f"{stat}" for stat in memory_usage_langfuse_items][
1525                    :top_k_items
1526                ],
1527                "total_usage": f"{total_memory_usage:.2f} MB",
1528                "langfuse_queue_length": self.task_manager._ingestion_queue.qsize(),
1529            }
1530
1531            self.log.debug("Memory usage: ", logged_memory_usage)
1532
1533            event = SdkLogBody(log=logged_memory_usage)
1534            self.task_manager.add_task(
1535                {
1536                    "id": str(uuid.uuid4()),
1537                    "type": "sdk-log",
1538                    "timestamp": _get_timestamp(),
1539                    "body": event.dict(),
1540                }
1541            )
1542
1543        except Exception as e:
1544            self.log.exception(e)
1545
1546    @overload
1547    def score(
1548        self,
1549        *,
1550        name: str,
1551        value: float,
1552        data_type: typing.Optional[Literal["NUMERIC", "BOOLEAN"]] = None,
1553        trace_id: typing.Optional[str] = None,
1554        id: typing.Optional[str] = None,
1555        comment: typing.Optional[str] = None,
1556        observation_id: typing.Optional[str] = None,
1557        config_id: typing.Optional[str] = None,
1558        **kwargs,
1559    ) -> "StatefulClient": ...
1560
1561    @overload
1562    def score(
1563        self,
1564        *,
1565        name: str,
1566        value: str,
1567        data_type: typing.Optional[Literal["CATEGORICAL"]] = "CATEGORICAL",
1568        trace_id: typing.Optional[str] = None,
1569        id: typing.Optional[str] = None,
1570        comment: typing.Optional[str] = None,
1571        observation_id: typing.Optional[str] = None,
1572        config_id: typing.Optional[str] = None,
1573        **kwargs,
1574    ) -> "StatefulClient": ...
1575
1576    def score(
1577        self,
1578        *,
1579        name: str,
1580        value: typing.Union[float, str],
1581        data_type: typing.Optional[ScoreDataType] = None,
1582        trace_id: typing.Optional[str] = None,
1583        id: typing.Optional[str] = None,
1584        comment: typing.Optional[str] = None,
1585        observation_id: typing.Optional[str] = None,
1586        config_id: typing.Optional[str] = None,
1587        **kwargs,
1588    ) -> "StatefulClient":
1589        """Create a score attached to a trace (and optionally an observation).
1590
1591        Args:
1592            name (str): Identifier of the score.
1593            value (Union[float, str]): The value of the score. Should be passed as float for numeric and boolean scores and as string for categorical scores.
1594            data_type (Optional[ScoreDataType]): The data type of the score. When not set, the data type is inferred from the score config's data type, when present.
1595              When no config is set, the data type is inferred from the value's type, i.e. float values are categorized as numeric scores and string values as categorical scores.
1596            trace_id (str): The id of the trace to which the score should be attached.
1597            id (Optional[str]): The id of the score. If not provided, a new UUID is generated.
1598            comment (Optional[str]): Additional context/explanation of the score.
1599            observation_id (Optional[str]): The id of the observation to which the score should be attached.
1600            config_id (Optional[str]): The id of the score config. When set, the score value is validated against the config. Defaults to None.
1601            **kwargs: Additional keyword arguments to include in the score.
1602
1603        Returns:
1604            StatefulClient: Either the associated observation (if observation_id is provided) or the trace (if observation_id is not provided).
1605
1606        Example:
1607            ```python
1608            from langfuse import Langfuse
1609
1610            langfuse = Langfuse()
1611
1612            # Create a trace
1613            trace = langfuse.trace(name="example-application")
1614
1615            # Get id of created trace
1616            trace_id = trace.id
1617
1618            # Add score to the trace
1619            trace = langfuse.score(
1620                trace_id=trace_id,
1621                name="user-explicit-feedback",
1622                value=0.9,
1623                comment="I like how personalized the response is"
1624            )
1625            ```
1626        """
1627        trace_id = trace_id or self.trace_id or str(uuid.uuid4())
1628        new_id = id or str(uuid.uuid4())
1629        try:
1630            new_dict = {
1631                "id": new_id,
1632                "trace_id": trace_id,
1633                "observation_id": observation_id,
1634                "name": name,
1635                "value": value,
1636                "data_type": data_type,
1637                "comment": comment,
1638                "config_id": config_id,
1639                **kwargs,
1640            }
1641
1642            self.log.debug(f"Creating score {new_dict}...")
1643            new_body = ScoreBody(**new_dict)
1644
1645            event = {
1646                "id": str(uuid.uuid4()),
1647                "type": "score-create",
1648                "body": new_body,
1649            }
1650            self.task_manager.add_task(event)
1651
1652        except Exception as e:
1653            self.log.exception(e)
1654        finally:
1655            if observation_id is not None:
1656                return StatefulClient(
1657                    self.client,
1658                    observation_id,
1659                    StateType.OBSERVATION,
1660                    trace_id,
1661                    self.task_manager,
1662                )
1663            else:
1664                return StatefulClient(
1665                    self.client, new_id, StateType.TRACE, new_id, self.task_manager
1666                )
1667
1668    def span(
1669        self,
1670        *,
1671        id: typing.Optional[str] = None,
1672        trace_id: typing.Optional[str] = None,
1673        parent_observation_id: typing.Optional[str] = None,
1674        name: typing.Optional[str] = None,
1675        start_time: typing.Optional[dt.datetime] = None,
1676        end_time: typing.Optional[dt.datetime] = None,
1677        metadata: typing.Optional[typing.Any] = None,
1678        level: typing.Optional[SpanLevel] = None,
1679        status_message: typing.Optional[str] = None,
1680        input: typing.Optional[typing.Any] = None,
1681        output: typing.Optional[typing.Any] = None,
1682        version: typing.Optional[str] = None,
1683        **kwargs,
1684    ) -> "StatefulSpanClient":
1685        """Create a span.
1686
1687        A span represents durations of units of work in a trace.
1688        Usually, you want to add a span nested within a trace. Optionally you can nest it within another observation by providing a parent_observation_id.
1689
1690        If no trace_id is provided, a new trace is created just for this span.
1691
1692        Args:
1693            id (Optional[str]): The id of the span can be set, otherwise a random id is generated. Spans are upserted on id.
1694            trace_id (Optional[str]): The trace ID associated with this span. If not provided, a new UUID is generated.
1695            parent_observation_id (Optional[str]): The ID of the parent observation, if applicable.
1696            name (Optional[str]): Identifier of the span. Useful for sorting/filtering in the UI.
1697            start_time (Optional[datetime]): The time at which the span started, defaults to the current time.
1698            end_time (Optional[datetime]): The time at which the span ended. Automatically set by `span.end()`.
1699            metadata (Optional[dict]): Additional metadata of the span. Can be any JSON object. Metadata is merged when being updated via the API.
1700            level (Optional[Literal["DEBUG", "DEFAULT", "WARNING", "ERROR"]]): The level of the span. Can be `DEBUG`, `DEFAULT`, `WARNING` or `ERROR`. Used for sorting/filtering of traces with elevated error levels and for highlighting in the UI.
1701            status_message (Optional[str]): The status message of the span. Additional field for context of the event. E.g. the error message of an error event.
1702            input (Optional[dict]): The input to the span. Can be any JSON object.
1703            output (Optional[dict]): The output to the span. Can be any JSON object.
1704            version (Optional[str]): The version of the span type. Used to understand how changes to the span type affect metrics. Useful in debugging.
1705            **kwargs: Additional keyword arguments to include in the span.
1706
1707        Returns:
1708            StatefulSpanClient: The created span.
1709
1710        Example:
1711            ```python
1712            from langfuse import Langfuse
1713
1714            langfuse = Langfuse()
1715
1716            trace = langfuse.trace(name = "llm-feature")
1717
1718            # Create a span
1719            retrieval = langfuse.span(name = "retrieval", trace_id = trace.id)
1720
1721            # Create a nested span
1722            nested_span = langfuse.span(name = "retrieval", trace_id = trace.id, parent_observation_id = retrieval.id)
1723            ```
1724        """
1725        new_span_id = id or str(uuid.uuid4())
1726        new_trace_id = trace_id or str(uuid.uuid4())
1727        self.trace_id = new_trace_id
1728        try:
1729            span_body = {
1730                "id": new_span_id,
1731                "trace_id": new_trace_id,
1732                "name": name,
1733                "start_time": start_time or _get_timestamp(),
1734                "metadata": metadata,
1735                "input": input,
1736                "output": output,
1737                "level": level,
1738                "status_message": status_message,
1739                "parent_observation_id": parent_observation_id,
1740                "version": version,
1741                "end_time": end_time,
1742                "trace": {"release": self.release},
1743                **kwargs,
1744            }
1745
1746            if trace_id is None:
1747                self._generate_trace(new_trace_id, name or new_trace_id)
1748
1749            self.log.debug(f"Creating span {_filter_io_from_event_body(span_body)}...")
1750
1751            span_body = CreateSpanBody(**span_body)
1752
1753            event = {
1754                "id": str(uuid.uuid4()),
1755                "type": "span-create",
1756                "body": span_body,
1757            }
1758
1759            self.task_manager.add_task(event)
1760
1761        except Exception as e:
1762            self.log.exception(e)
1763        finally:
1764            self._log_memory_usage()
1765
1766            return StatefulSpanClient(
1767                self.client,
1768                new_span_id,
1769                StateType.OBSERVATION,
1770                new_trace_id,
1771                self.task_manager,
1772            )
1773
1774    def event(
1775        self,
1776        *,
1777        id: typing.Optional[str] = None,
1778        trace_id: typing.Optional[str] = None,
1779        parent_observation_id: typing.Optional[str] = None,
1780        name: typing.Optional[str] = None,
1781        start_time: typing.Optional[dt.datetime] = None,
1782        metadata: typing.Optional[typing.Any] = None,
1783        input: typing.Optional[typing.Any] = None,
1784        output: typing.Optional[typing.Any] = None,
1785        level: typing.Optional[SpanLevel] = None,
1786        status_message: typing.Optional[str] = None,
1787        version: typing.Optional[str] = None,
1788        **kwargs,
1789    ) -> "StatefulSpanClient":
1790        """Create an event.
1791
1792        An event represents a discrete event in a trace.
1793        Usually, you want to add a event nested within a trace. Optionally you can nest it within another observation by providing a parent_observation_id.
1794
1795        If no trace_id is provided, a new trace is created just for this event.
1796
1797        Args:
1798            id (Optional[str]): The id of the event can be set, otherwise a random id is generated.
1799            trace_id (Optional[str]): The trace ID associated with this event. If not provided, a new trace is created just for this event.
1800            parent_observation_id (Optional[str]): The ID of the parent observation, if applicable.
1801            name (Optional[str]): Identifier of the event. Useful for sorting/filtering in the UI.
1802            start_time (Optional[datetime]): The time at which the event started, defaults to the current time.
1803            metadata (Optional[Any]): Additional metadata of the event. Can be any JSON object. Metadata is merged when being updated via the API.
1804            input (Optional[Any]): The input to the event. Can be any JSON object.
1805            output (Optional[Any]): The output to the event. Can be any JSON object.
1806            level (Optional[Literal["DEBUG", "DEFAULT", "WARNING", "ERROR"]]): The level of the event. Can be `DEBUG`, `DEFAULT`, `WARNING` or `ERROR`. Used for sorting/filtering of traces with elevated error levels and for highlighting in the UI.
1807            status_message (Optional[str]): The status message of the event. Additional field for context of the event. E.g. the error message of an error event.
1808            version (Optional[str]): The version of the event type. Used to understand how changes to the event type affect metrics. Useful in debugging.
1809            **kwargs: Additional keyword arguments to include in the event.
1810
1811        Returns:
1812            StatefulSpanClient: The created event.
1813
1814        Example:
1815            ```python
1816            from langfuse import Langfuse
1817
1818            langfuse = Langfuse()
1819
1820            trace = langfuse.trace(name = "llm-feature")
1821
1822            # Create an event
1823            retrieval = langfuse.event(name = "retrieval", trace_id = trace.id)
1824            ```
1825        """
1826        event_id = id or str(uuid.uuid4())
1827        new_trace_id = trace_id or str(uuid.uuid4())
1828        self.trace_id = new_trace_id
1829        try:
1830            event_body = {
1831                "id": event_id,
1832                "trace_id": new_trace_id,
1833                "name": name,
1834                "start_time": start_time or _get_timestamp(),
1835                "metadata": metadata,
1836                "input": input,
1837                "output": output,
1838                "level": level,
1839                "status_message": status_message,
1840                "parent_observation_id": parent_observation_id,
1841                "version": version,
1842                "trace": {"release": self.release},
1843                **kwargs,
1844            }
1845
1846            if trace_id is None:
1847                self._generate_trace(new_trace_id, name or new_trace_id)
1848
1849            request = CreateEventBody(**event_body)
1850
1851            event = {
1852                "id": str(uuid.uuid4()),
1853                "type": "event-create",
1854                "body": request,
1855            }
1856
1857            self.log.debug(
1858                f"Creating event {_filter_io_from_event_body(event_body)} ..."
1859            )
1860            self.task_manager.add_task(event)
1861
1862        except Exception as e:
1863            self.log.exception(e)
1864        finally:
1865            return StatefulSpanClient(
1866                self.client,
1867                event_id,
1868                StateType.OBSERVATION,
1869                new_trace_id,
1870                self.task_manager,
1871            )
1872
1873    def generation(
1874        self,
1875        *,
1876        id: typing.Optional[str] = None,
1877        trace_id: typing.Optional[str] = None,
1878        parent_observation_id: typing.Optional[str] = None,
1879        name: typing.Optional[str] = None,
1880        start_time: typing.Optional[dt.datetime] = None,
1881        end_time: typing.Optional[dt.datetime] = None,
1882        completion_start_time: typing.Optional[dt.datetime] = None,
1883        metadata: typing.Optional[typing.Any] = None,
1884        level: typing.Optional[SpanLevel] = None,
1885        status_message: typing.Optional[str] = None,
1886        version: typing.Optional[str] = None,
1887        model: typing.Optional[str] = None,
1888        model_parameters: typing.Optional[typing.Dict[str, MapValue]] = None,
1889        input: typing.Optional[typing.Any] = None,
1890        output: typing.Optional[typing.Any] = None,
1891        usage: typing.Optional[typing.Union[pydantic.BaseModel, ModelUsage]] = None,
1892        usage_details: typing.Optional[typing.Dict[str, int]] = None,
1893        cost_details: typing.Optional[typing.Dict[str, float]] = None,
1894        prompt: typing.Optional[PromptClient] = None,
1895        **kwargs,
1896    ) -> "StatefulGenerationClient":
1897        """Create a generation.
1898
1899        A generation is a span that is used to log generations of AI models. They contain additional metadata about the model, the prompt/completion, the cost of executing the model and are specifically rendered in the langfuse UI.
1900
1901        Usually, you want to add a generation nested within a trace. Optionally you can nest it within another observation by providing a parent_observation_id.
1902
1903        If no trace_id is provided, a new trace is created just for this generation.
1904
1905        Args:
1906            id (Optional[str]): The id of the generation can be set, defaults to random id.
1907            trace_id (Optional[str]): The trace ID associated with this generation. If not provided, a new trace is created
1908            parent_observation_id (Optional[str]): The ID of the parent observation, if applicable.
1909            name (Optional[str]): Identifier of the generation. Useful for sorting/filtering in the UI.
1910            start_time (Optional[datetime.datetime]): The time at which the generation started, defaults to the current time.
1911            end_time (Optional[datetime.datetime]): The time at which the generation ended. Automatically set by `generation.end()`.
1912            completion_start_time (Optional[datetime.datetime]): The time at which the completion started (streaming). Set it to get latency analytics broken down into time until completion started and completion duration.
1913            metadata (Optional[dict]): Additional metadata of the generation. Can be any JSON object. Metadata is merged when being updated via the API.
1914            level (Optional[str]): The level of the generation. Can be `DEBUG`, `DEFAULT`, `WARNING` or `ERROR`. Used for sorting/filtering of traces with elevated error levels and for highlighting in the UI.
1915            status_message (Optional[str]): The status message of the generation. Additional field for context of the event. E.g. the error message of an error event.
1916            version (Optional[str]): The version of the generation type. Used to understand how changes to the span type affect metrics. Useful in debugging.
1917            model (Optional[str]): The name of the model used for the generation.
1918            model_parameters (Optional[dict]): The parameters of the model used for the generation; can be any key-value pairs.
1919            input (Optional[dict]): The prompt used for the generation. Can be any string or JSON object.
1920            output (Optional[dict]): The completion generated by the model. Can be any string or JSON object.
1921            usage (Optional[dict]): [DEPRECATED, use usage_details and cost_details instead] The usage object supports the OpenAi structure with {`promptTokens`, `completionTokens`, `totalTokens`} and a more generic version {`input`, `output`, `total`, `unit`, `inputCost`, `outputCost`, `totalCost`} where unit can be of value `"TOKENS"`, `"CHARACTERS"`, `"MILLISECONDS"`, `"SECONDS"`, or `"IMAGES"`. Refer to the docs on how to [automatically infer](https://langfuse.com/docs/model-usage-and-cost) token usage and costs in Langfuse.
1922            usage_details (Optional[dict]): The usage details of the generation. Also accepts OpenAI usage details. Keys are the usage type (e.g. "input", "input_cached", "output") and values are integers representing the number of units used. For accurate cost calculations in Langfuse, ensure each usage type has a corresponding price configured in the Langfuse models table. Example: {"input": 500, "output": 150}.
1923            cost_details (Optional[dict]): The cost details of the generation. Keys are the usage type (e.g. "input", "input_cached", "output") and values are floats representing the cost in USD. Example: {"input": 0.0015, "output": 0.002}.
1924            prompt (Optional[PromptClient]): The Langfuse prompt object used for the generation.
1925            **kwargs: Additional keyword arguments to include in the generation.
1926
1927        Returns:
1928            StatefulGenerationClient: The created generation.
1929
1930        Example:
1931            ```python
1932            from langfuse import Langfuse
1933
1934            langfuse = Langfuse()
1935
1936            # Create a generation in Langfuse
1937            generation = langfuse.generation(
1938                name="summary-generation",
1939                model="gpt-3.5-turbo",
1940                model_parameters={"maxTokens": "1000", "temperature": "0.9"},
1941                input=[{"role": "system", "content": "You are a helpful assistant."},
1942                       {"role": "user", "content": "Please generate a summary of the following documents ..."}],
1943                metadata={"interface": "whatsapp"}
1944            )
1945            ```
1946        """
1947        new_trace_id = trace_id or str(uuid.uuid4())
1948        new_generation_id = id or str(uuid.uuid4())
1949        self.trace_id = new_trace_id
1950        try:
1951            generation_body = {
1952                "id": new_generation_id,
1953                "trace_id": new_trace_id,
1954                "release": self.release,
1955                "name": name,
1956                "start_time": start_time or _get_timestamp(),
1957                "metadata": metadata,
1958                "input": input,
1959                "output": output,
1960                "level": level,
1961                "status_message": status_message,
1962                "parent_observation_id": parent_observation_id,
1963                "version": version,
1964                "end_time": end_time,
1965                "completion_start_time": completion_start_time,
1966                "model": model,
1967                "model_parameters": model_parameters,
1968                "usage": _convert_usage_input(usage) if usage is not None else None,
1969                "usage_details": usage_details,
1970                "cost_details": cost_details,
1971                "trace": {"release": self.release},
1972                **_create_prompt_context(prompt),
1973                **kwargs,
1974            }
1975
1976            if trace_id is None:
1977                trace = {
1978                    "id": new_trace_id,
1979                    "release": self.release,
1980                    "name": name,
1981                }
1982                request = TraceBody(**trace)
1983
1984                event = {
1985                    "id": str(uuid.uuid4()),
1986                    "type": "trace-create",
1987                    "body": request,
1988                }
1989
1990                self.log.debug("Creating trace...")
1991
1992                self.task_manager.add_task(event)
1993
1994            self.log.debug(
1995                f"Creating generation max {_filter_io_from_event_body(generation_body)}..."
1996            )
1997            request = CreateGenerationBody(**generation_body)
1998
1999            event = {
2000                "id": str(uuid.uuid4()),
2001                "type": "generation-create",
2002                "body": request,
2003            }
2004
2005            self.task_manager.add_task(event)
2006
2007        except Exception as e:
2008            self.log.exception(e)
2009        finally:
2010            return StatefulGenerationClient(
2011                self.client,
2012                new_generation_id,
2013                StateType.OBSERVATION,
2014                new_trace_id,
2015                self.task_manager,
2016            )
2017
2018    def _generate_trace(self, trace_id: str, name: str):
2019        trace_dict = {
2020            "id": trace_id,
2021            "release": self.release,
2022            "name": name,
2023        }
2024
2025        trace_body = TraceBody(**trace_dict)
2026
2027        event = {
2028            "id": str(uuid.uuid4()),
2029            "type": "trace-create",
2030            "body": trace_body,
2031        }
2032
2033        self.log.debug(f"Creating trace {_filter_io_from_event_body(trace_dict)}...")
2034        self.task_manager.add_task(event)
2035
2036    def join(self):
2037        """Blocks until all consumer Threads are terminated. The SKD calls this upon termination of the Python Interpreter.
2038
2039        If called before flushing, consumers might terminate before sending all events to Langfuse API. This method is called at exit of the SKD, right before the Python interpreter closes.
2040        To guarantee all messages have been delivered, you still need to call flush().
2041        """
2042        try:
2043            return self.task_manager.join()
2044        except Exception as e:
2045            self.log.exception(e)
2046
2047    def flush(self):
2048        """Flush the internal event queue to the Langfuse API. It blocks until the queue is empty. It should be called when the application shuts down.
2049
2050        Example:
2051            ```python
2052            from langfuse import Langfuse
2053
2054            langfuse = Langfuse()
2055
2056            # Some operations with Langfuse
2057
2058            # Flushing all events to end Langfuse cleanly
2059            langfuse.flush()
2060            ```
2061        """
2062        try:
2063            return self.task_manager.flush()
2064        except Exception as e:
2065            self.log.exception(e)
2066
2067    def shutdown(self):
2068        """Initiate a graceful shutdown of the Langfuse SDK, ensuring all events are sent to Langfuse API and all consumer Threads are terminated.
2069
2070        This function calls flush() and join() consecutively resulting in a complete shutdown of the SDK. On success of this function, no more events will be sent to Langfuse API.
2071        As the SDK calls join() already on shutdown, refer to flush() to ensure all events arive at the Langfuse API.
2072        """
2073        try:
2074            self.prompt_cache._task_manager.shutdown()
2075
2076            # In logging.py, a handler is attached to the httpx logger.
2077            # To avoid a memory leak on singleton reset, remove all handlers
2078            httpx_logger = logging.getLogger("httpx")
2079            for handler in httpx_logger.handlers:
2080                httpx_logger.removeHandler(handler)
2081
2082            return self.task_manager.shutdown()
2083        except Exception as e:
2084            self.log.exception(e)
2085
2086
2087class StateType(Enum):
2088    """Enum to distinguish observation and trace states.
2089
2090    Attributes:
2091        OBSERVATION (int): Observation state.
2092        TRACE (int): Trace state.
2093    """
2094
2095    OBSERVATION = 1
2096    TRACE = 0
2097
2098
2099class StatefulClient(object):
2100    """Base class for handling stateful operations in the Langfuse system.
2101
2102    This client is capable of creating different nested Langfuse objects like spans, generations, scores, and events,
2103    associating them with either an observation or a trace based on the specified state type.
2104
2105    Attributes:
2106        client (FernLangfuse): Core interface for Langfuse API interactions.
2107        id (str): Unique identifier of the stateful client (either observation or trace).
2108        state_type (StateType): Enum indicating whether the client is an observation or a trace.
2109        trace_id (str): Id of the trace associated with the stateful client.
2110        task_manager (TaskManager): Manager handling asynchronous tasks for the client.
2111    """
2112
2113    log = logging.getLogger("langfuse")
2114
2115    def __init__(
2116        self,
2117        client: FernLangfuse,
2118        id: str,
2119        state_type: StateType,
2120        trace_id: str,
2121        task_manager: TaskManager,
2122    ):
2123        """Initialize the StatefulClient.
2124
2125        Args:
2126            client (FernLangfuse): Core interface for Langfuse API interactions.
2127            id (str): Unique identifier of the stateful client (either observation or trace).
2128            state_type (StateType): Enum indicating whether the client is an observation or a trace.
2129            trace_id (str): Id of the trace associated with the stateful client.
2130            task_manager (TaskManager): Manager handling asynchronous tasks for the client.
2131        """
2132        self.client = client
2133        self.trace_id = trace_id
2134        self.id = id
2135        self.state_type = state_type
2136        self.task_manager = task_manager
2137
2138    def _add_state_to_event(self, body: dict):
2139        if self.state_type == StateType.OBSERVATION:
2140            body["parent_observation_id"] = self.id
2141            body["trace_id"] = self.trace_id
2142        else:
2143            body["trace_id"] = self.id
2144        return body
2145
2146    def _add_default_values(self, body: dict):
2147        if body.get("start_time") is None:
2148            body["start_time"] = _get_timestamp()
2149        return body
2150
2151    def generation(
2152        self,
2153        *,
2154        id: typing.Optional[str] = None,
2155        name: typing.Optional[str] = None,
2156        start_time: typing.Optional[dt.datetime] = None,
2157        end_time: typing.Optional[dt.datetime] = None,
2158        metadata: typing.Optional[typing.Any] = None,
2159        level: typing.Optional[SpanLevel] = None,
2160        status_message: typing.Optional[str] = None,
2161        version: typing.Optional[str] = None,
2162        completion_start_time: typing.Optional[dt.datetime] = None,
2163        model: typing.Optional[str] = None,
2164        model_parameters: typing.Optional[typing.Dict[str, MapValue]] = None,
2165        input: typing.Optional[typing.Any] = None,
2166        output: typing.Optional[typing.Any] = None,
2167        usage: typing.Optional[typing.Union[pydantic.BaseModel, ModelUsage]] = None,
2168        usage_details: typing.Optional[typing.Dict[str, int]] = None,
2169        cost_details: typing.Optional[typing.Dict[str, float]] = None,
2170        prompt: typing.Optional[PromptClient] = None,
2171        **kwargs,
2172    ) -> "StatefulGenerationClient":
2173        """Create a generation nested within the current observation or trace.
2174
2175        A generation is a span that is used to log generations of AI models. They contain additional metadata about the model, the prompt/completion, the cost of executing the model and are specifically rendered in the langfuse UI.
2176
2177        Args:
2178            id (Optional[str]): The id of the generation can be set, defaults to random id.
2179            name (Optional[str]): Identifier of the generation. Useful for sorting/filtering in the UI.
2180            start_time (Optional[datetime.datetime]): The time at which the generation started, defaults to the current time.
2181            end_time (Optional[datetime.datetime]): The time at which the generation ended. Automatically set by `generation.end()`.
2182            completion_start_time (Optional[datetime.datetime]): The time at which the completion started (streaming). Set it to get latency analytics broken down into time until completion started and completion duration.
2183            metadata (Optional[dict]): Additional metadata of the generation. Can be any JSON object. Metadata is merged when being updated via the API.
2184            level (Optional[str]): The level of the generation. Can be `DEBUG`, `DEFAULT`, `WARNING` or `ERROR`. Used for sorting/filtering of traces with elevated error levels and for highlighting in the UI.
2185            status_message (Optional[str]): The status message of the generation. Additional field for context of the event. E.g. the error message of an error event.
2186            version (Optional[str]): The version of the generation type. Used to understand how changes to the span type affect metrics. Useful in debugging.
2187            model (Optional[str]): The name of the model used for the generation.
2188            model_parameters (Optional[dict]): The parameters of the model used for the generation; can be any key-value pairs.
2189            input (Optional[dict]): The prompt used for the generation. Can be any string or JSON object.
2190            output (Optional[dict]): The completion generated by the model. Can be any string or JSON object.
2191            usage (Optional[dict]): [DEPRECATED, use usage_details and cost_details instead] The usage object supports the OpenAi structure with {`promptTokens`, `completionTokens`, `totalTokens`} and a more generic version {`input`, `output`, `total`, `unit`, `inputCost`, `outputCost`, `totalCost`} where unit can be of value `"TOKENS"`, `"CHARACTERS"`, `"MILLISECONDS"`, `"SECONDS"`, or `"IMAGES"`. Refer to the docs on how to [automatically infer](https://langfuse.com/docs/model-usage-and-cost) token usage and costs in Langfuse.
2192            usage_details (Optional[dict]): The usage details of the generation. Also accepts OpenAI usage details. Keys are the usage type (e.g. "input", "input_cached", "output") and values are integers representing the number of units used. For accurate cost calculations in Langfuse, ensure each usage type has a corresponding price configured in the Langfuse models table. Example: {"input": 500, "output": 150}.
2193            cost_details (Optional[dict]): The cost details of the generation. Keys are the usage type (e.g. "input", "input_cached", "output") and values are floats representing the cost in USD. Example: {"input": 0.0015, "output": 0.002}.
2194            prompt (Optional[PromptClient]): The Langfuse prompt object used for the generation.
2195            **kwargs: Additional keyword arguments to include in the generation.
2196
2197        Returns:
2198            StatefulGenerationClient: The created generation. Use this client to update the generation or create additional nested observations.
2199
2200        Example:
2201            ```python
2202            from langfuse import Langfuse
2203
2204            langfuse = Langfuse()
2205
2206            # Create a trace
2207            trace = langfuse.trace(name = "llm-feature")
2208
2209            # Create a nested generation in Langfuse
2210            generation = trace.generation(
2211                name="summary-generation",
2212                model="gpt-3.5-turbo",
2213                model_parameters={"maxTokens": "1000", "temperature": "0.9"},
2214                input=[{"role": "system", "content": "You are a helpful assistant."},
2215                       {"role": "user", "content": "Please generate a summary of the following documents ..."}],
2216                metadata={"interface": "whatsapp"}
2217            )
2218            ```
2219        """
2220        generation_id = id or str(uuid.uuid4())
2221        try:
2222            generation_body = {
2223                "id": generation_id,
2224                "name": name,
2225                "start_time": start_time or _get_timestamp(),
2226                "metadata": metadata,
2227                "level": level,
2228                "status_message": status_message,
2229                "version": version,
2230                "end_time": end_time,
2231                "completion_start_time": completion_start_time,
2232                "model": model,
2233                "model_parameters": model_parameters,
2234                "input": input,
2235                "output": output,
2236                "usage": _convert_usage_input(usage) if usage is not None else None,
2237                "usage_details": usage_details,
2238                "cost_details": cost_details,
2239                **_create_prompt_context(prompt),
2240                **kwargs,
2241            }
2242
2243            generation_body = self._add_state_to_event(generation_body)
2244            new_body = self._add_default_values(generation_body)
2245
2246            new_body = CreateGenerationBody(**new_body)
2247
2248            event = {
2249                "id": str(uuid.uuid4()),
2250                "type": "generation-create",
2251                "body": new_body.dict(exclude_none=True, exclude_unset=False),
2252            }
2253
2254            self.log.debug(
2255                f"Creating generation {_filter_io_from_event_body(generation_body)}..."
2256            )
2257            self.task_manager.add_task(event)
2258
2259        except Exception as e:
2260            self.log.exception(e)
2261        finally:
2262            return StatefulGenerationClient(
2263                self.client,
2264                generation_id,
2265                StateType.OBSERVATION,
2266                self.trace_id,
2267                task_manager=self.task_manager,
2268            )
2269
2270    def span(
2271        self,
2272        *,
2273        id: typing.Optional[str] = None,
2274        name: typing.Optional[str] = None,
2275        start_time: typing.Optional[dt.datetime] = None,
2276        end_time: typing.Optional[dt.datetime] = None,
2277        metadata: typing.Optional[typing.Any] = None,
2278        input: typing.Optional[typing.Any] = None,
2279        output: typing.Optional[typing.Any] = None,
2280        level: typing.Optional[SpanLevel] = None,
2281        status_message: typing.Optional[str] = None,
2282        version: typing.Optional[str] = None,
2283        **kwargs,
2284    ) -> "StatefulSpanClient":
2285        """Create a span nested within the current observation or trace.
2286
2287        A span represents durations of units of work in a trace.
2288
2289        Args:
2290            id (Optional[str]): The id of the span can be set, otherwise a random id is generated. Spans are upserted on id.
2291            name (Optional[str]): Identifier of the span. Useful for sorting/filtering in the UI.
2292            start_time (Optional[datetime]): The time at which the span started, defaults to the current time.
2293            end_time (Optional[datetime]): The time at which the span ended. Automatically set by `span.end()`.
2294            metadata (Optional[dict]): Additional metadata of the span. Can be any JSON object. Metadata is merged when being updated via the API.
2295            level (Optional[Literal["DEBUG", "DEFAULT", "WARNING", "ERROR"]]): The level of the span. Can be `DEBUG`, `DEFAULT`, `WARNING` or `ERROR`. Used for sorting/filtering of traces with elevated error levels and for highlighting in the UI.
2296            status_message (Optional[str]): The status message of the span. Additional field for context of the event. E.g. the error message of an error event.
2297            input (Optional[dict]): The input to the span. Can be any JSON object.
2298            output (Optional[dict]): The output to the span. Can be any JSON object.
2299            version (Optional[str]): The version of the span type. Used to understand how changes to the span type affect metrics. Useful in debugging.
2300            **kwargs: Additional keyword arguments to include in the span.
2301
2302        Returns:
2303            StatefulSpanClient: The created span. Use this client to update the span or create additional nested observations.
2304
2305        Example:
2306            ```python
2307            from langfuse import Langfuse
2308
2309            langfuse = Langfuse()
2310
2311            # Create a trace
2312            trace = langfuse.trace(name = "llm-feature")
2313
2314            # Create a span
2315            retrieval = langfuse.span(name = "retrieval")
2316            ```
2317        """
2318        span_id = id or str(uuid.uuid4())
2319        try:
2320            span_body = {
2321                "id": span_id,
2322                "name": name,
2323                "start_time": start_time or _get_timestamp(),
2324                "metadata": metadata,
2325                "input": input,
2326                "output": output,
2327                "level": level,
2328                "status_message": status_message,
2329                "version": version,
2330                "end_time": end_time,
2331                **kwargs,
2332            }
2333
2334            self.log.debug(f"Creating span {_filter_io_from_event_body(span_body)}...")
2335
2336            new_dict = self._add_state_to_event(span_body)
2337            new_body = self._add_default_values(new_dict)
2338
2339            event = CreateSpanBody(**new_body)
2340
2341            event = {
2342                "id": str(uuid.uuid4()),
2343                "type": "span-create",
2344                "body": event,
2345            }
2346
2347            self.task_manager.add_task(event)
2348        except Exception as e:
2349            self.log.exception(e)
2350        finally:
2351            return StatefulSpanClient(
2352                self.client,
2353                span_id,
2354                StateType.OBSERVATION,
2355                self.trace_id,
2356                task_manager=self.task_manager,
2357            )
2358
2359    @overload
2360    def score(
2361        self,
2362        *,
2363        id: typing.Optional[str] = None,
2364        name: str,
2365        value: float,
2366        data_type: typing.Optional[Literal["NUMERIC", "BOOLEAN"]] = None,
2367        comment: typing.Optional[str] = None,
2368        config_id: typing.Optional[str] = None,
2369        **kwargs,
2370    ) -> "StatefulClient": ...
2371
2372    @overload
2373    def score(
2374        self,
2375        *,
2376        id: typing.Optional[str] = None,
2377        name: str,
2378        value: str,
2379        data_type: typing.Optional[Literal["CATEGORICAL"]] = "CATEGORICAL",
2380        comment: typing.Optional[str] = None,
2381        config_id: typing.Optional[str] = None,
2382        **kwargs,
2383    ) -> "StatefulClient": ...
2384
2385    def score(
2386        self,
2387        *,
2388        id: typing.Optional[str] = None,
2389        name: str,
2390        value: typing.Union[float, str],
2391        data_type: typing.Optional[ScoreDataType] = None,
2392        comment: typing.Optional[str] = None,
2393        config_id: typing.Optional[str] = None,
2394        **kwargs,
2395    ) -> "StatefulClient":
2396        """Create a score attached for the current observation or trace.
2397
2398        Args:
2399            name (str): Identifier of the score.
2400            value (Union[float, str]): The value of the score. Should be passed as float for numeric and boolean scores and as string for categorical scores.
2401            data_type (Optional[ScoreDataType]): The data type of the score. When not set, the data type is inferred from the score config's data type, when present.
2402              When no config is set, the data type is inferred from the value's type, i.e. float values are categorized as numeric scores and string values as categorical scores.
2403            comment (Optional[str]): Additional context/explanation of the score.
2404            id (Optional[str]): The id of the score. If not provided, a new UUID is generated.
2405            config_id (Optional[str]): The id of the score config. When set, the score value is validated against the config. Defaults to None.
2406            **kwargs: Additional keyword arguments to include in the score.
2407
2408        Returns:
2409            StatefulClient: The current observation or trace for which the score was created. Passthrough for chaining.
2410
2411        Example:
2412            ```python
2413            from langfuse import Langfuse
2414
2415            langfuse = Langfuse()
2416
2417            # Create a trace
2418            trace = langfuse.trace(name="example-application")
2419
2420            # Add score to the trace
2421            trace = trace.score(
2422                name="user-explicit-feedback",
2423                value=0.8,
2424                comment="I like how personalized the response is"
2425            )
2426            ```
2427        """
2428        score_id = id or str(uuid.uuid4())
2429        try:
2430            new_score = {
2431                "id": score_id,
2432                "trace_id": self.trace_id,
2433                "name": name,
2434                "value": value,
2435                "data_type": data_type,
2436                "comment": comment,
2437                "config_id": config_id,
2438                **kwargs,
2439            }
2440
2441            self.log.debug(f"Creating score {new_score}...")
2442
2443            new_dict = self._add_state_to_event(new_score)
2444
2445            if self.state_type == StateType.OBSERVATION:
2446                new_dict["observationId"] = self.id
2447
2448            request = ScoreBody(**new_dict)
2449
2450            event = {
2451                "id": str(uuid.uuid4()),
2452                "type": "score-create",
2453                "body": request,
2454            }
2455
2456            self.task_manager.add_task(event)
2457
2458        except Exception as e:
2459            self.log.exception(e)
2460        finally:
2461            return StatefulClient(
2462                self.client,
2463                self.id,
2464                self.state_type,
2465                self.trace_id,
2466                task_manager=self.task_manager,
2467            )
2468
2469    def event(
2470        self,
2471        *,
2472        id: typing.Optional[str] = None,
2473        name: typing.Optional[str] = None,
2474        start_time: typing.Optional[dt.datetime] = None,
2475        metadata: typing.Optional[typing.Any] = None,
2476        input: typing.Optional[typing.Any] = None,
2477        output: typing.Optional[typing.Any] = None,
2478        level: typing.Optional[SpanLevel] = None,
2479        status_message: typing.Optional[str] = None,
2480        version: typing.Optional[str] = None,
2481        **kwargs,
2482    ) -> "StatefulClient":
2483        """Create an event nested within the current observation or trace.
2484
2485        An event represents a discrete event in a trace.
2486
2487        Args:
2488            id (Optional[str]): The id of the event can be set, otherwise a random id is generated.
2489            name (Optional[str]): Identifier of the event. Useful for sorting/filtering in the UI.
2490            start_time (Optional[datetime]): The time at which the event started, defaults to the current time.
2491            metadata (Optional[Any]): Additional metadata of the event. Can be any JSON object. Metadata is merged when being updated via the API.
2492            input (Optional[Any]): The input to the event. Can be any JSON object.
2493            output (Optional[Any]): The output to the event. Can be any JSON object.
2494            level (Optional[Literal["DEBUG", "DEFAULT", "WARNING", "ERROR"]]): The level of the event. Can be `DEBUG`, `DEFAULT`, `WARNING` or `ERROR`. Used for sorting/filtering of traces with elevated error levels and for highlighting in the UI.
2495            status_message (Optional[str]): The status message of the event. Additional field for context of the event. E.g. the error message of an error event.
2496            version (Optional[str]): The version of the event type. Used to understand how changes to the event type affect metrics. Useful in debugging.
2497            **kwargs: Additional keyword arguments to include in the event.
2498
2499        Returns:
2500            StatefulSpanClient: The created event. Use this client to update the event or create additional nested observations.
2501
2502        Example:
2503            ```python
2504            from langfuse import Langfuse
2505
2506            langfuse = Langfuse()
2507
2508            # Create a trace
2509            trace = langfuse.trace(name = "llm-feature")
2510
2511            # Create an event
2512            retrieval = trace.event(name = "retrieval")
2513            ```
2514        """
2515        event_id = id or str(uuid.uuid4())
2516        try:
2517            event_body = {
2518                "id": event_id,
2519                "name": name,
2520                "start_time": start_time or _get_timestamp(),
2521                "metadata": metadata,
2522                "input": input,
2523                "output": output,
2524                "level": level,
2525                "status_message": status_message,
2526                "version": version,
2527                **kwargs,
2528            }
2529
2530            new_dict = self._add_state_to_event(event_body)
2531            new_body = self._add_default_values(new_dict)
2532
2533            request = CreateEventBody(**new_body)
2534
2535            event = {
2536                "id": str(uuid.uuid4()),
2537                "type": "event-create",
2538                "body": request,
2539            }
2540
2541            self.log.debug(
2542                f"Creating event {_filter_io_from_event_body(event_body)}..."
2543            )
2544            self.task_manager.add_task(event)
2545
2546        except Exception as e:
2547            self.log.exception(e)
2548        finally:
2549            return StatefulClient(
2550                self.client,
2551                event_id,
2552                StateType.OBSERVATION,
2553                self.trace_id,
2554                self.task_manager,
2555            )
2556
2557    def get_trace_url(self):
2558        """Get the URL to see the current trace in the Langfuse UI."""
2559        return f"{self.client._client_wrapper._base_url}/trace/{self.trace_id}"
2560
2561
2562class StatefulGenerationClient(StatefulClient):
2563    """Class for handling stateful operations of generations in the Langfuse system. Inherits from StatefulClient.
2564
2565    This client extends the capabilities of the StatefulClient to specifically handle generation,
2566    allowing for the creation, update, and termination of generation processes in Langfuse.
2567
2568    Attributes:
2569        client (FernLangfuse): Core interface for Langfuse API interaction.
2570        id (str): Unique identifier of the generation.
2571        state_type (StateType): Type of the stateful entity (observation or trace).
2572        trace_id (str): Id of trace associated with the generation.
2573        task_manager (TaskManager): Manager for handling asynchronous tasks.
2574    """
2575
2576    log = logging.getLogger("langfuse")
2577
2578    def __init__(
2579        self,
2580        client: FernLangfuse,
2581        id: str,
2582        state_type: StateType,
2583        trace_id: str,
2584        task_manager: TaskManager,
2585    ):
2586        """Initialize the StatefulGenerationClient."""
2587        super().__init__(client, id, state_type, trace_id, task_manager)
2588
2589    # WHEN CHANGING THIS METHOD, UPDATE END() FUNCTION ACCORDINGLY
2590    def update(
2591        self,
2592        *,
2593        name: typing.Optional[str] = None,
2594        start_time: typing.Optional[dt.datetime] = None,
2595        end_time: typing.Optional[dt.datetime] = None,
2596        completion_start_time: typing.Optional[dt.datetime] = None,
2597        metadata: typing.Optional[typing.Any] = None,
2598        level: typing.Optional[SpanLevel] = None,
2599        status_message: typing.Optional[str] = None,
2600        version: typing.Optional[str] = None,
2601        model: typing.Optional[str] = None,
2602        model_parameters: typing.Optional[typing.Dict[str, MapValue]] = None,
2603        input: typing.Optional[typing.Any] = None,
2604        output: typing.Optional[typing.Any] = None,
2605        usage: typing.Optional[typing.Union[pydantic.BaseModel, ModelUsage]] = None,
2606        usage_details: typing.Optional[typing.Dict[str, int]] = None,
2607        cost_details: typing.Optional[typing.Dict[str, float]] = None,
2608        prompt: typing.Optional[PromptClient] = None,
2609        **kwargs,
2610    ) -> "StatefulGenerationClient":
2611        """Update the generation.
2612
2613        Args:
2614            name (Optional[str]): Identifier of the generation. Useful for sorting/filtering in the UI.
2615            start_time (Optional[datetime.datetime]): The time at which the generation started, defaults to the current time.
2616            end_time (Optional[datetime.datetime]): The time at which the generation ended. Automatically set by `generation.end()`.
2617            completion_start_time (Optional[datetime.datetime]): The time at which the completion started (streaming). Set it to get latency analytics broken down into time until completion started and completion duration.
2618            metadata (Optional[dict]): Additional metadata of the generation. Can be any JSON object. Metadata is merged when being updated via the API.
2619            level (Optional[str]): The level of the generation. Can be `DEBUG`, `DEFAULT`, `WARNING` or `ERROR`. Used for sorting/filtering of traces with elevated error levels and for highlighting in the UI.
2620            status_message (Optional[str]): The status message of the generation. Additional field for context of the event. E.g. the error message of an error event.
2621            version (Optional[str]): The version of the generation type. Used to understand how changes to the span type affect metrics. Useful in debugging.
2622            model (Optional[str]): The name of the model used for the generation.
2623            model_parameters (Optional[dict]): The parameters of the model used for the generation; can be any key-value pairs.
2624            input (Optional[dict]): The prompt used for the generation. Can be any string or JSON object.
2625            output (Optional[dict]): The completion generated by the model. Can be any string or JSON object.
2626            usage (Optional[dict]): [DEPRECATED, use usage_details and cost_details instead] The usage object supports the OpenAi structure with {`promptTokens`, `completionTokens`, `totalTokens`} and a more generic version {`input`, `output`, `total`, `unit`, `inputCost`, `outputCost`, `totalCost`} where unit can be of value `"TOKENS"`, `"CHARACTERS"`, `"MILLISECONDS"`, `"SECONDS"`, or `"IMAGES"`. Refer to the docs on how to [automatically infer](https://langfuse.com/docs/model-usage-and-cost) token usage and costs in Langfuse.
2627            usage_details (Optional[dict]): The usage details of the generation. Also accepts OpenAI usage details. Keys are the usage type (e.g. "input", "input_cached", "output") and values are integers representing the number of units used. For accurate cost calculations in Langfuse, ensure each usage type has a corresponding price configured in the Langfuse models table. Example: {"input": 500, "output": 150}.
2628            cost_details (Optional[dict]): The cost details of the generation. Keys are the usage type (e.g. "input", "input_cached", "output") and values are floats representing the cost in USD. Example: {"input": 0.0015, "output": 0.002}.
2629            prompt (Optional[PromptClient]): The Langfuse prompt object used for the generation.
2630            **kwargs: Additional keyword arguments to include in the generation.
2631
2632        Returns:
2633            StatefulGenerationClient: The updated generation. Passthrough for chaining.
2634
2635        Example:
2636            ```python
2637            from langfuse import Langfuse
2638
2639            langfuse = Langfuse()
2640
2641            # Create a trace
2642            trace = langfuse.trace(name = "llm-feature")
2643
2644            # Create a nested generation in Langfuse
2645            generation = trace.generation(name="summary-generation")
2646
2647            # Update the generation
2648            generation = generation.update(metadata={"interface": "whatsapp"})
2649            ```
2650        """
2651        try:
2652            generation_body = {
2653                "id": self.id,
2654                "trace_id": self.trace_id,  # Included to avoid relying on the order of events sent to the API
2655                "name": name,
2656                "start_time": start_time,
2657                "metadata": metadata,
2658                "level": level,
2659                "status_message": status_message,
2660                "version": version,
2661                "end_time": end_time,
2662                "completion_start_time": completion_start_time,
2663                "model": model,
2664                "model_parameters": model_parameters,
2665                "input": input,
2666                "output": output,
2667                "usage": _convert_usage_input(usage) if usage is not None else None,
2668                "usage_details": usage_details,
2669                "cost_details": cost_details,
2670                **_create_prompt_context(prompt),
2671                **kwargs,
2672            }
2673
2674            self.log.debug(
2675                f"Update generation {_filter_io_from_event_body(generation_body)}..."
2676            )
2677
2678            request = UpdateGenerationBody(**generation_body)
2679
2680            event = {
2681                "id": str(uuid.uuid4()),
2682                "type": "generation-update",
2683                "body": request.dict(exclude_none=True, exclude_unset=False),
2684            }
2685
2686            self.log.debug(
2687                f"Update generation {_filter_io_from_event_body(generation_body)}..."
2688            )
2689            self.task_manager.add_task(event)
2690
2691        except Exception as e:
2692            self.log.exception(e)
2693        finally:
2694            return StatefulGenerationClient(
2695                self.client,
2696                self.id,
2697                StateType.OBSERVATION,
2698                self.trace_id,
2699                task_manager=self.task_manager,
2700            )
2701
2702    def end(
2703        self,
2704        *,
2705        name: typing.Optional[str] = None,
2706        start_time: typing.Optional[dt.datetime] = None,
2707        end_time: typing.Optional[dt.datetime] = None,
2708        completion_start_time: typing.Optional[dt.datetime] = None,
2709        metadata: typing.Optional[typing.Any] = None,
2710        level: typing.Optional[SpanLevel] = None,
2711        status_message: typing.Optional[str] = None,
2712        version: typing.Optional[str] = None,
2713        model: typing.Optional[str] = None,
2714        model_parameters: typing.Optional[typing.Dict[str, MapValue]] = None,
2715        input: typing.Optional[typing.Any] = None,
2716        output: typing.Optional[typing.Any] = None,
2717        usage: typing.Optional[typing.Union[pydantic.BaseModel, ModelUsage]] = None,
2718        usage_details: typing.Optional[typing.Dict[str, int]] = None,
2719        cost_details: typing.Optional[typing.Dict[str, float]] = None,
2720        prompt: typing.Optional[PromptClient] = None,
2721        **kwargs,
2722    ) -> "StatefulGenerationClient":
2723        """End the generation, optionally updating its properties.
2724
2725        Args:
2726            name (Optional[str]): Identifier of the generation. Useful for sorting/filtering in the UI.
2727            start_time (Optional[datetime.datetime]): The time at which the generation started, defaults to the current time.
2728            end_time (Optional[datetime.datetime]): Automatically set to the current time. Can be overridden to set a custom end time.
2729            completion_start_time (Optional[datetime.datetime]): The time at which the completion started (streaming). Set it to get latency analytics broken down into time until completion started and completion duration.
2730            metadata (Optional[dict]): Additional metadata of the generation. Can be any JSON object. Metadata is merged when being updated via the API.
2731            level (Optional[str]): The level of the generation. Can be `DEBUG`, `DEFAULT`, `WARNING` or `ERROR`. Used for sorting/filtering of traces with elevated error levels and for highlighting in the UI.
2732            status_message (Optional[str]): The status message of the generation. Additional field for context of the event. E.g. the error message of an error event.
2733            version (Optional[str]): The version of the generation type. Used to understand how changes to the span type affect metrics. Useful in debugging.
2734            model (Optional[str]): The name of the model used for the generation.
2735            model_parameters (Optional[dict]): The parameters of the model used for the generation; can be any key-value pairs.
2736            input (Optional[dict]): The prompt used for the generation. Can be any string or JSON object.
2737            output (Optional[dict]): The completion generated by the model. Can be any string or JSON object.
2738            usage (Optional[dict]): [DEPRECATED, use usage_details and cost_details instead] The usage object supports the OpenAi structure with {`promptTokens`, `completionTokens`, `totalTokens`} and a more generic version {`input`, `output`, `total`, `unit`, `inputCost`, `outputCost`, `totalCost`} where unit can be of value `"TOKENS"`, `"CHARACTERS"`, `"MILLISECONDS"`, `"SECONDS"`, or `"IMAGES"`. Refer to the docs on how to [automatically infer](https://langfuse.com/docs/model-usage-and-cost) token usage and costs in Langfuse.
2739            usage_details (Optional[dict]): The usage details of the generation. Also accepts OpenAI usage details. Keys are the usage type (e.g. "input", "input_cached", "output") and values are integers representing the number of units used. For accurate cost calculations in Langfuse, ensure each usage type has a corresponding price configured in the Langfuse models table. Example: {"input": 500, "output": 150}.
2740            cost_details (Optional[dict]): The cost details of the generation. Keys are the usage type (e.g. "input", "input_cached", "output") and values are floats representing the cost in USD. Example: {"input": 0.0015, "output": 0.002}.
2741            prompt (Optional[PromptClient]): The Langfuse prompt object used for the generation.
2742            **kwargs: Additional keyword arguments to include in the generation.
2743
2744        Returns:
2745            StatefulGenerationClient: The ended generation. Passthrough for chaining.
2746
2747        Example:
2748            ```python
2749            from langfuse import Langfuse
2750
2751            langfuse = Langfuse()
2752
2753            # Create a trace
2754            trace = langfuse.trace(name = "llm-feature")
2755
2756            # Create a nested generation in Langfuse
2757            generation = trace.generation(name="summary-generation")
2758
2759            # End the generation and update its properties
2760            generation = generation.end(metadata={"interface": "whatsapp"})
2761            ```
2762        """
2763        return self.update(
2764            name=name,
2765            start_time=start_time,
2766            end_time=end_time or _get_timestamp(),
2767            metadata=metadata,
2768            level=level,
2769            status_message=status_message,
2770            version=version,
2771            completion_start_time=completion_start_time,
2772            model=model,
2773            model_parameters=model_parameters,
2774            input=input,
2775            output=output,
2776            usage=usage,
2777            usage_details=usage_details,
2778            cost_details=cost_details,
2779            prompt=prompt,
2780            **kwargs,
2781        )
2782
2783
2784class StatefulSpanClient(StatefulClient):
2785    """Class for handling stateful operations of spans in the Langfuse system. Inherits from StatefulClient.
2786
2787    Attributes:
2788        client (FernLangfuse): Core interface for Langfuse API interaction.
2789        id (str): Unique identifier of the span.
2790        state_type (StateType): Type of the stateful entity (observation or trace).
2791        trace_id (str): Id of trace associated with the span.
2792        task_manager (TaskManager): Manager for handling asynchronous tasks.
2793    """
2794
2795    log = logging.getLogger("langfuse")
2796
2797    def __init__(
2798        self,
2799        client: FernLangfuse,
2800        id: str,
2801        state_type: StateType,
2802        trace_id: str,
2803        task_manager: TaskManager,
2804    ):
2805        """Initialize the StatefulSpanClient."""
2806        super().__init__(client, id, state_type, trace_id, task_manager)
2807
2808    # WHEN CHANGING THIS METHOD, UPDATE END() FUNCTION ACCORDINGLY
2809    def update(
2810        self,
2811        *,
2812        name: typing.Optional[str] = None,
2813        start_time: typing.Optional[dt.datetime] = None,
2814        end_time: typing.Optional[dt.datetime] = None,
2815        metadata: typing.Optional[typing.Any] = None,
2816        input: typing.Optional[typing.Any] = None,
2817        output: typing.Optional[typing.Any] = None,
2818        level: typing.Optional[SpanLevel] = None,
2819        status_message: typing.Optional[str] = None,
2820        version: typing.Optional[str] = None,
2821        **kwargs,
2822    ) -> "StatefulSpanClient":
2823        """Update the span.
2824
2825        Args:
2826            name (Optional[str]): Identifier of the span. Useful for sorting/filtering in the UI.
2827            start_time (Optional[datetime]): The time at which the span started, defaults to the current time.
2828            end_time (Optional[datetime]): The time at which the span ended. Automatically set by `span.end()`.
2829            metadata (Optional[dict]): Additional metadata of the span. Can be any JSON object. Metadata is merged when being updated via the API.
2830            level (Optional[Literal["DEBUG", "DEFAULT", "WARNING", "ERROR"]]): The level of the span. Can be `DEBUG`, `DEFAULT`, `WARNING` or `ERROR`. Used for sorting/filtering of traces with elevated error levels and for highlighting in the UI.
2831            status_message (Optional[str]): The status message of the span. Additional field for context of the event. E.g. the error message of an error event.
2832            input (Optional[dict]): The input to the span. Can be any JSON object.
2833            output (Optional[dict]): The output to the span. Can be any JSON object.
2834            version (Optional[str]): The version of the span type. Used to understand how changes to the span type affect metrics. Useful in debugging.
2835            **kwargs: Additional keyword arguments to include in the span.
2836
2837        Returns:
2838            StatefulSpanClient: The updated span. Passthrough for chaining.
2839
2840        Example:
2841            ```python
2842            from langfuse import Langfuse
2843
2844            langfuse = Langfuse()
2845
2846            # Create a trace
2847            trace = langfuse.trace(name = "llm-feature")
2848
2849            # Create a nested span in Langfuse
2850            span = trace.span(name="retrieval")
2851
2852            # Update the span
2853            span = span.update(metadata={"interface": "whatsapp"})
2854            ```
2855        """
2856        try:
2857            span_body = {
2858                "id": self.id,
2859                "trace_id": self.trace_id,  # Included to avoid relying on the order of events sent to the API
2860                "name": name,
2861                "start_time": start_time,
2862                "metadata": metadata,
2863                "input": input,
2864                "output": output,
2865                "level": level,
2866                "status_message": status_message,
2867                "version": version,
2868                "end_time": end_time,
2869                **kwargs,
2870            }
2871            self.log.debug(f"Update span {_filter_io_from_event_body(span_body)}...")
2872
2873            request = UpdateSpanBody(**span_body)
2874
2875            event = {
2876                "id": str(uuid.uuid4()),
2877                "type": "span-update",
2878                "body": request,
2879            }
2880
2881            self.task_manager.add_task(event)
2882        except Exception as e:
2883            self.log.exception(e)
2884        finally:
2885            return StatefulSpanClient(
2886                self.client,
2887                self.id,
2888                StateType.OBSERVATION,
2889                self.trace_id,
2890                task_manager=self.task_manager,
2891            )
2892
2893    def end(
2894        self,
2895        *,
2896        name: typing.Optional[str] = None,
2897        start_time: typing.Optional[dt.datetime] = None,
2898        end_time: typing.Optional[dt.datetime] = None,
2899        metadata: typing.Optional[typing.Any] = None,
2900        input: typing.Optional[typing.Any] = None,
2901        output: typing.Optional[typing.Any] = None,
2902        level: typing.Optional[SpanLevel] = None,
2903        status_message: typing.Optional[str] = None,
2904        version: typing.Optional[str] = None,
2905        **kwargs,
2906    ) -> "StatefulSpanClient":
2907        """End the span, optionally updating its properties.
2908
2909        Args:
2910            name (Optional[str]): Identifier of the span. Useful for sorting/filtering in the UI.
2911            start_time (Optional[datetime]): The time at which the span started, defaults to the current time.
2912            end_time (Optional[datetime]): The time at which the span ended. Automatically set by `span.end()`.
2913            metadata (Optional[dict]): Additional metadata of the span. Can be any JSON object. Metadata is merged when being updated via the API.
2914            level (Optional[Literal["DEBUG", "DEFAULT", "WARNING", "ERROR"]]): The level of the span. Can be `DEBUG`, `DEFAULT`, `WARNING` or `ERROR`. Used for sorting/filtering of traces with elevated error levels and for highlighting in the UI.
2915            status_message (Optional[str]): The status message of the span. Additional field for context of the event. E.g. the error message of an error event.
2916            input (Optional[dict]): The input to the span. Can be any JSON object.
2917            output (Optional[dict]): The output to the span. Can be any JSON object.
2918            version (Optional[str]): The version of the span type. Used to understand how changes to the span type affect metrics. Useful in debugging.
2919            **kwargs: Additional keyword arguments to include in the span.
2920
2921        Returns:
2922            StatefulSpanClient: The updated span. Passthrough for chaining.
2923
2924        Example:
2925            ```python
2926            from langfuse import Langfuse
2927
2928            langfuse = Langfuse()
2929
2930            # Create a trace
2931            trace = langfuse.trace(name = "llm-feature")
2932
2933            # Create a nested span in Langfuse
2934            span = trace.span(name="retrieval")
2935
2936            # End the span and update its properties
2937            span = span.end(metadata={"interface": "whatsapp"})
2938            ```
2939        """
2940        try:
2941            span_body = {
2942                "name": name,
2943                "start_time": start_time,
2944                "metadata": metadata,
2945                "input": input,
2946                "output": output,
2947                "level": level,
2948                "status_message": status_message,
2949                "version": version,
2950                "end_time": end_time or _get_timestamp(),
2951                **kwargs,
2952            }
2953            return self.update(**span_body)
2954
2955        except Exception as e:
2956            self.log.warning(e)
2957        finally:
2958            return StatefulSpanClient(
2959                self.client,
2960                self.id,
2961                StateType.OBSERVATION,
2962                self.trace_id,
2963                task_manager=self.task_manager,
2964            )
2965
2966    def get_langchain_handler(self, update_parent: bool = False):
2967        """Get langchain callback handler associated with the current span.
2968
2969        Args:
2970            update_parent (bool): If set to True, the parent observation will be updated with the outcome of the Langchain run.
2971
2972        Returns:
2973            CallbackHandler: An instance of CallbackHandler linked to this StatefulSpanClient.
2974        """
2975        from langfuse.callback import CallbackHandler
2976
2977        return CallbackHandler(
2978            stateful_client=self, update_stateful_client=update_parent
2979        )
2980
2981
2982class StatefulTraceClient(StatefulClient):
2983    """Class for handling stateful operations of traces in the Langfuse system. Inherits from StatefulClient.
2984
2985    Attributes:
2986        client (FernLangfuse): Core interface for Langfuse API interaction.
2987        id (str): Unique identifier of the trace.
2988        state_type (StateType): Type of the stateful entity (observation or trace).
2989        trace_id (str): The trace ID associated with this client.
2990        task_manager (TaskManager): Manager for handling asynchronous tasks.
2991    """
2992
2993    log = logging.getLogger("langfuse")
2994
2995    def __init__(
2996        self,
2997        client: FernLangfuse,
2998        id: str,
2999        state_type: StateType,
3000        trace_id: str,
3001        task_manager: TaskManager,
3002    ):
3003        """Initialize the StatefulTraceClient."""
3004        super().__init__(client, id, state_type, trace_id, task_manager)
3005        self.task_manager = task_manager
3006
3007    def update(
3008        self,
3009        *,
3010        name: typing.Optional[str] = None,
3011        user_id: typing.Optional[str] = None,
3012        session_id: typing.Optional[str] = None,
3013        version: typing.Optional[str] = None,
3014        release: typing.Optional[str] = None,
3015        input: typing.Optional[typing.Any] = None,
3016        output: typing.Optional[typing.Any] = None,
3017        metadata: typing.Optional[typing.Any] = None,
3018        tags: typing.Optional[typing.List[str]] = None,
3019        public: typing.Optional[bool] = None,
3020        **kwargs,
3021    ) -> "StatefulTraceClient":
3022        """Update the trace.
3023
3024        Args:
3025            name: Identifier of the trace. Useful for sorting/filtering in the UI.
3026            input: The input of the trace. Can be any JSON object.
3027            output: The output of the trace. Can be any JSON object.
3028            metadata: Additional metadata of the trace. Can be any JSON object. Metadata is merged when being updated via the API.
3029            user_id: The id of the user that triggered the execution. Used to provide user-level analytics.
3030            session_id: Used to group multiple traces into a session in Langfuse. Use your own session/thread identifier.
3031            version: The version of the trace type. Used to understand how changes to the trace type affect metrics. Useful in debugging.
3032            release: The release identifier of the current deployment. Used to understand how changes of different deployments affect metrics. Useful in debugging.
3033            tags: Tags are used to categorize or label traces. Traces can be filtered by tags in the UI and GET API. Tags can also be changed in the UI. Tags are merged and never deleted via the API.
3034            public: You can make a trace `public` to share it via a public link. This allows others to view the trace without needing to log in or be members of your Langfuse project.
3035            **kwargs: Additional keyword arguments that can be included in the trace.
3036
3037        Returns:
3038            StatefulTraceClient: The updated trace. Passthrough for chaining.
3039
3040        Example:
3041            ```python
3042            from langfuse import Langfuse
3043
3044            langfuse = Langfuse()
3045
3046            # Create a trace
3047            trace = langfuse.trace(
3048                name="example-application",
3049                user_id="user-1234")
3050            )
3051
3052            # Update the trace
3053            trace = trace.update(
3054                output={"result": "success"},
3055                metadata={"interface": "whatsapp"}
3056            )
3057            ```
3058        """
3059        try:
3060            trace_body = {
3061                "id": self.id,
3062                "name": name,
3063                "userId": user_id,
3064                "sessionId": session_id
3065                or kwargs.get("sessionId", None),  # backward compatibility
3066                "version": version,
3067                "release": release,
3068                "input": input,
3069                "output": output,
3070                "metadata": metadata,
3071                "public": public,
3072                "tags": tags,
3073                **kwargs,
3074            }
3075            self.log.debug(f"Update trace {_filter_io_from_event_body(trace_body)}...")
3076
3077            request = TraceBody(**trace_body)
3078
3079            event = {
3080                "id": str(uuid.uuid4()),
3081                "type": "trace-create",
3082                "body": request,
3083            }
3084
3085            self.task_manager.add_task(event)
3086
3087        except Exception as e:
3088            self.log.exception(e)
3089        finally:
3090            return StatefulTraceClient(
3091                self.client,
3092                self.id,
3093                StateType.TRACE,
3094                self.trace_id,
3095                task_manager=self.task_manager,
3096            )
3097
3098    def get_langchain_handler(self, update_parent: bool = False):
3099        """Get langchain callback handler associated with the current trace.
3100
3101        This method creates and returns a CallbackHandler instance, linking it with the current
3102        trace. Use this if you want to group multiple Langchain runs within a single trace.
3103
3104        Args:
3105            update_parent (bool): If set to True, the parent trace will be updated with the outcome of the Langchain run.
3106
3107        Raises:
3108            ImportError: If the 'langchain' module is not installed, indicating missing functionality.
3109
3110        Returns:
3111            CallbackHandler: Langchain callback handler linked to the current trace.
3112
3113        Example:
3114            ```python
3115            from langfuse import Langfuse
3116
3117            langfuse = Langfuse()
3118
3119            # Create a trace
3120            trace = langfuse.trace(name = "llm-feature")
3121
3122            # Get a langchain callback handler
3123            handler = trace.get_langchain_handler()
3124            ```
3125        """
3126        try:
3127            from langfuse.callback import CallbackHandler
3128
3129            self.log.debug(f"Creating new handler for trace {self.id}")
3130
3131            return CallbackHandler(
3132                stateful_client=self,
3133                debug=self.log.level == logging.DEBUG,
3134                update_stateful_client=update_parent,
3135            )
3136        except Exception as e:
3137            self.log.exception(e)
3138
3139    def getNewHandler(self):
3140        """Alias for the `get_langchain_handler` method. Retrieves a callback handler for the trace. Deprecated."""
3141        return self.get_langchain_handler()
3142
3143
3144class DatasetItemClient:
3145    """Class for managing dataset items in Langfuse.
3146
3147    Args:
3148        id (str): Unique identifier of the dataset item.
3149        status (DatasetStatus): The status of the dataset item. Can be either 'ACTIVE' or 'ARCHIVED'.
3150        input (Any): Input data of the dataset item.
3151        expected_output (Optional[Any]): Expected output of the dataset item.
3152        metadata (Optional[Any]): Additional metadata of the dataset item.
3153        source_trace_id (Optional[str]): Identifier of the source trace.
3154        source_observation_id (Optional[str]): Identifier of the source observation.
3155        dataset_id (str): Identifier of the dataset to which this item belongs.
3156        dataset_name (str): Name of the dataset to which this item belongs.
3157        created_at (datetime): Timestamp of dataset item creation.
3158        updated_at (datetime): Timestamp of the last update to the dataset item.
3159        langfuse (Langfuse): Instance of Langfuse client for API interactions.
3160
3161    Example:
3162        ```python
3163        from langfuse import Langfuse
3164
3165        langfuse = Langfuse()
3166
3167        dataset = langfuse.get_dataset("<dataset_name>")
3168
3169        for item in dataset.items:
3170            # Generate a completion using the input of every item
3171            completion, generation = llm_app.run(item.input)
3172
3173            # Evaluate the completion
3174            generation.score(
3175                name="example-score",
3176                value=1
3177            )
3178        ```
3179    """
3180
3181    log = logging.getLogger("langfuse")
3182
3183    id: str
3184    status: DatasetStatus
3185    input: typing.Any
3186    expected_output: typing.Optional[typing.Any]
3187    metadata: Optional[Any]
3188    source_trace_id: typing.Optional[str]
3189    source_observation_id: typing.Optional[str]
3190    dataset_id: str
3191    dataset_name: str
3192    created_at: dt.datetime
3193    updated_at: dt.datetime
3194
3195    langfuse: Langfuse
3196
3197    def __init__(self, dataset_item: DatasetItem, langfuse: Langfuse):
3198        """Initialize the DatasetItemClient."""
3199        self.id = dataset_item.id
3200        self.status = dataset_item.status
3201        self.input = dataset_item.input
3202        self.expected_output = dataset_item.expected_output
3203        self.metadata = dataset_item.metadata
3204        self.source_trace_id = dataset_item.source_trace_id
3205        self.source_observation_id = dataset_item.source_observation_id
3206        self.dataset_id = dataset_item.dataset_id
3207        self.dataset_name = dataset_item.dataset_name
3208        self.created_at = dataset_item.created_at
3209        self.updated_at = dataset_item.updated_at
3210
3211        self.langfuse = langfuse
3212
3213    def flush(self, observation: StatefulClient, run_name: str):
3214        """Flushes an observations task manager's queue.
3215
3216        Used before creating a dataset run item to ensure all events are persistent.
3217
3218        Args:
3219            observation (StatefulClient): The observation or trace client associated with the dataset item.
3220            run_name (str): The name of the dataset run.
3221        """
3222        observation.task_manager.flush()
3223
3224    def link(
3225        self,
3226        trace_or_observation: typing.Union[StatefulClient, str, None],
3227        run_name: str,
3228        run_metadata: Optional[Any] = None,
3229        run_description: Optional[str] = None,
3230        trace_id: Optional[str] = None,
3231        observation_id: Optional[str] = None,
3232    ):
3233        """Link the dataset item to observation within a specific dataset run. Creates a dataset run item.
3234
3235        Args:
3236            trace_or_observation (Union[StatefulClient, str, None]): The trace or observation object to link. Deprecated: can also be an observation ID.
3237            run_name (str): The name of the dataset run.
3238            run_metadata (Optional[Any]): Additional metadata to include in dataset run.
3239            run_description (Optional[str]): Description of the dataset run.
3240            trace_id (Optional[str]): The trace ID to link to the dataset item. Set trace_or_observation to None if trace_id is provided.
3241            observation_id (Optional[str]): The observation ID to link to the dataset item (optional). Set trace_or_observation to None if trace_id is provided.
3242        """
3243        parsed_trace_id: str = None
3244        parsed_observation_id: str = None
3245
3246        if isinstance(trace_or_observation, StatefulClient):
3247            # flush the queue before creating the dataset run item
3248            # to ensure that all events are persisted.
3249            if trace_or_observation.state_type == StateType.TRACE:
3250                parsed_trace_id = trace_or_observation.trace_id
3251            elif trace_or_observation.state_type == StateType.OBSERVATION:
3252                parsed_observation_id = trace_or_observation.id
3253                parsed_trace_id = trace_or_observation.trace_id
3254        # legacy support for observation_id
3255        elif isinstance(trace_or_observation, str):
3256            parsed_observation_id = trace_or_observation
3257        elif trace_or_observation is None:
3258            if trace_id is not None:
3259                parsed_trace_id = trace_id
3260                if observation_id is not None:
3261                    parsed_observation_id = observation_id
3262            else:
3263                raise ValueError(
3264                    "trace_id must be provided if trace_or_observation is None"
3265                )
3266        else:
3267            raise ValueError(
3268                "trace_or_observation (arg) or trace_id (kwarg) must be provided to link the dataset item"
3269            )
3270
3271        self.log.debug(
3272            f"Creating dataset run item: {run_name} {self.id} {parsed_trace_id} {parsed_observation_id}"
3273        )
3274        self.langfuse.client.dataset_run_items.create(
3275            request=CreateDatasetRunItemRequest(
3276                runName=run_name,
3277                datasetItemId=self.id,
3278                traceId=parsed_trace_id,
3279                observationId=parsed_observation_id,
3280                metadata=run_metadata,
3281                runDescription=run_description,
3282            )
3283        )
3284
3285    def get_langchain_handler(
3286        self,
3287        *,
3288        run_name: str,
3289        run_description: Optional[str] = None,
3290        run_metadata: Optional[Any] = None,
3291    ):
3292        """Create and get a langchain callback handler linked to this dataset item.
3293
3294        Args:
3295            run_name (str): The name of the dataset run to be used in the callback handler.
3296            run_description (Optional[str]): Description of the dataset run.
3297            run_metadata (Optional[Any]): Additional metadata to include in dataset run.
3298
3299        Returns:
3300            CallbackHandler: An instance of CallbackHandler linked to the dataset item.
3301        """
3302        metadata = {
3303            "dataset_item_id": self.id,
3304            "run_name": run_name,
3305            "dataset_id": self.dataset_id,
3306        }
3307        trace = self.langfuse.trace(name="dataset-run", metadata=metadata)
3308
3309        self.link(
3310            trace, run_name, run_metadata=run_metadata, run_description=run_description
3311        )
3312
3313        return trace.get_langchain_handler(update_parent=True)
3314
3315    @contextmanager
3316    def observe(
3317        self,
3318        *,
3319        run_name: str,
3320        run_description: Optional[str] = None,
3321        run_metadata: Optional[Any] = None,
3322        trace_id: Optional[str] = None,
3323    ):
3324        """Observes a dataset run within the Langfuse client.
3325
3326        Args:
3327            run_name (str): The name of the dataset run.
3328            root_trace (Optional[StatefulTraceClient]): The root trace client to use for the dataset run. If not provided, a new trace client will be created.
3329            run_description (Optional[str]): The description of the dataset run.
3330            run_metadata (Optional[Any]): Additional metadata for the dataset run.
3331
3332        Yields:
3333            StatefulTraceClient: The trace associated with the dataset run.
3334        """
3335        from langfuse.decorators import langfuse_context
3336
3337        root_trace_id = trace_id or str(uuid.uuid4())
3338
3339        langfuse_context._set_root_trace_id(root_trace_id)
3340
3341        try:
3342            yield root_trace_id
3343
3344        finally:
3345            self.link(
3346                run_name=run_name,
3347                run_metadata=run_metadata,
3348                run_description=run_description,
3349                trace_or_observation=None,
3350                trace_id=root_trace_id,
3351            )
3352
3353    @contextmanager
3354    def observe_llama_index(
3355        self,
3356        *,
3357        run_name: str,
3358        run_description: Optional[str] = None,
3359        run_metadata: Optional[Any] = None,
3360        llama_index_integration_constructor_kwargs: Optional[Dict[str, Any]] = {},
3361    ):
3362        """Context manager for observing LlamaIndex operations linked to this dataset item.
3363
3364        This method sets up a LlamaIndex callback handler that integrates with Langfuse, allowing detailed logging
3365        and tracing of LlamaIndex operations within the context of a specific dataset run. It ensures that all
3366        operations performed within the context are linked to the appropriate dataset item and run in Langfuse.
3367
3368        Args:
3369            run_name (str): The name of the dataset run.
3370            run_description (Optional[str]): Description of the dataset run. Defaults to None.
3371            run_metadata (Optional[Any]): Additional metadata for the dataset run. Defaults to None.
3372            llama_index_integration_constructor_kwargs (Optional[Dict[str, Any]]): Keyword arguments to pass
3373                to the LlamaIndex integration constructor. Defaults to an empty dictionary.
3374
3375        Yields:
3376            LlamaIndexCallbackHandler: The callback handler for LlamaIndex operations.
3377
3378        Example:
3379            ```python
3380            dataset_item = dataset.items[0]
3381
3382            with dataset_item.observe_llama_index(run_name="example-run", run_description="Example LlamaIndex run") as handler:
3383                # Perform LlamaIndex operations here
3384                some_llama_index_operation()
3385            ```
3386
3387        Raises:
3388            ImportError: If required modules for LlamaIndex integration are not available.
3389        """
3390        metadata = {
3391            "dataset_item_id": self.id,
3392            "run_name": run_name,
3393            "dataset_id": self.dataset_id,
3394        }
3395        trace = self.langfuse.trace(name="dataset-run", metadata=metadata)
3396        self.link(
3397            trace, run_name, run_metadata=run_metadata, run_description=run_description
3398        )
3399
3400        try:
3401            import llama_index.core
3402            from llama_index.core import Settings
3403            from llama_index.core.callbacks import CallbackManager
3404
3405            from langfuse.llama_index import LlamaIndexCallbackHandler
3406
3407            callback_handler = LlamaIndexCallbackHandler(
3408                **llama_index_integration_constructor_kwargs,
3409            )
3410            callback_handler.set_root(trace, update_root=True)
3411
3412            # Temporarily set the global handler to the new handler if previous handler is a LlamaIndexCallbackHandler
3413            # LlamaIndex does not adding two errors of same type, so if global handler is already a LlamaIndexCallbackHandler, we need to remove it
3414            prev_global_handler = llama_index.core.global_handler
3415            prev_langfuse_handler = None
3416
3417            if isinstance(prev_global_handler, LlamaIndexCallbackHandler):
3418                llama_index.core.global_handler = None
3419
3420            if Settings.callback_manager is None:
3421                Settings.callback_manager = CallbackManager([callback_handler])
3422            else:
3423                for handler in Settings.callback_manager.handlers:
3424                    if isinstance(handler, LlamaIndexCallbackHandler):
3425                        prev_langfuse_handler = handler
3426                        Settings.callback_manager.remove_handler(handler)
3427
3428                Settings.callback_manager.add_handler(callback_handler)
3429
3430        except Exception as e:
3431            self.log.exception(e)
3432
3433        try:
3434            yield callback_handler
3435        finally:
3436            # Reset the handlers
3437            Settings.callback_manager.remove_handler(callback_handler)
3438            if prev_langfuse_handler is not None:
3439                Settings.callback_manager.add_handler(prev_langfuse_handler)
3440
3441            llama_index.core.global_handler = prev_global_handler
3442
3443    def get_llama_index_handler(
3444        self,
3445        *,
3446        run_name: str,
3447        run_description: Optional[str] = None,
3448        run_metadata: Optional[Any] = None,
3449        llama_index_integration_constructor_kwargs: Optional[Dict[str, Any]] = {},
3450    ):
3451        """Create and get a llama-index callback handler linked to this dataset item.
3452
3453        Args:
3454            run_name (str): The name of the dataset run to be used in the callback handler.
3455            run_description (Optional[str]): Description of the dataset run.
3456            run_metadata (Optional[Any]): Additional metadata to include in dataset run.
3457            llama_index_integration_constructor_kwargs (Optional[Dict[str, Any]]): Additional keyword arguments to pass to the LlamaIndex integration constructor.
3458
3459        Returns:
3460            LlamaIndexCallbackHandler: An instance of LlamaIndexCallbackHandler linked to the dataset item.
3461        """
3462        metadata = {
3463            "dataset_item_id": self.id,
3464            "run_name": run_name,
3465            "dataset_id": self.dataset_id,
3466        }
3467        trace = self.langfuse.trace(name="dataset-run", metadata=metadata)
3468
3469        self.link(
3470            trace, run_name, run_metadata=run_metadata, run_description=run_description
3471        )
3472
3473        try:
3474            from langfuse.llama_index.llama_index import LlamaIndexCallbackHandler
3475
3476            callback_handler = LlamaIndexCallbackHandler(
3477                **llama_index_integration_constructor_kwargs,
3478            )
3479            callback_handler.set_root(trace, update_root=True)
3480
3481            return callback_handler
3482        except Exception as e:
3483            self.log.exception(e)
3484
3485
3486class DatasetClient:
3487    """Class for managing datasets in Langfuse.
3488
3489    Attributes:
3490        id (str): Unique identifier of the dataset.
3491        name (str): Name of the dataset.
3492        description (Optional[str]): Description of the dataset.
3493        metadata (Optional[typing.Any]): Additional metadata of the dataset.
3494        project_id (str): Identifier of the project to which the dataset belongs.
3495        dataset_name (str): Name of the dataset.
3496        created_at (datetime): Timestamp of dataset creation.
3497        updated_at (datetime): Timestamp of the last update to the dataset.
3498        items (List[DatasetItemClient]): List of dataset items associated with the dataset.
3499        runs (List[str]): List of dataset runs associated with the dataset. Deprecated.
3500
3501    Example:
3502        Print the input of each dataset item in a dataset.
3503        ```python
3504        from langfuse import Langfuse
3505
3506        langfuse = Langfuse()
3507
3508        dataset = langfuse.get_dataset("<dataset_name>")
3509
3510        for item in dataset.items:
3511            print(item.input)
3512        ```
3513    """
3514
3515    id: str
3516    name: str
3517    description: Optional[str]
3518    project_id: str
3519    dataset_name: str  # for backward compatibility, to be deprecated
3520    metadata: Optional[Any]
3521    created_at: dt.datetime
3522    updated_at: dt.datetime
3523    items: typing.List[DatasetItemClient]
3524    runs: typing.List[str] = []  # deprecated
3525
3526    def __init__(self, dataset: Dataset, items: typing.List[DatasetItemClient]):
3527        """Initialize the DatasetClient."""
3528        self.id = dataset.id
3529        self.name = dataset.name
3530        self.description = dataset.description
3531        self.project_id = dataset.project_id
3532        self.metadata = dataset.metadata
3533        self.dataset_name = dataset.name  # for backward compatibility, to be deprecated
3534        self.created_at = dataset.created_at
3535        self.updated_at = dataset.updated_at
3536        self.items = items
3537
3538
3539def _filter_io_from_event_body(event_body: Dict[str, Any]):
3540    return {
3541        k: v for k, v in event_body.items() if k not in ("input", "output", "metadata")
3542    }