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