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