134class Langfuse:
 135    """Main client for Langfuse tracing and platform features.
 136
 137    This class provides an interface for creating and managing traces, spans,
 138    and generations in Langfuse as well as interacting with the Langfuse API.
 139
 140    The client features a thread-safe singleton pattern for each unique public API key,
 141    ensuring consistent trace context propagation across your application. It implements
 142    efficient batching of spans with configurable flush settings and includes background
 143    thread management for media uploads and score ingestion.
 144
 145    Configuration is flexible through either direct parameters or environment variables,
 146    with graceful fallbacks and runtime configuration updates.
 147
 148    Attributes:
 149        api: Synchronous API client for Langfuse backend communication
 150        async_api: Asynchronous API client for Langfuse backend communication
 151        _otel_tracer: Internal LangfuseTracer instance managing OpenTelemetry components
 152
 153    Parameters:
 154        public_key (Optional[str]): Your Langfuse public API key. Can also be set via LANGFUSE_PUBLIC_KEY environment variable.
 155        secret_key (Optional[str]): Your Langfuse secret API key. Can also be set via LANGFUSE_SECRET_KEY environment variable.
 156        base_url (Optional[str]): The Langfuse API base URL. Defaults to "https://cloud.langfuse.com". Can also be set via LANGFUSE_BASE_URL environment variable.
 157        host (Optional[str]): Deprecated. Use base_url instead. The Langfuse API host URL. Defaults to "https://cloud.langfuse.com".
 158        timeout (Optional[int]): Timeout in seconds for API requests. Defaults to 5 seconds.
 159        httpx_client (Optional[httpx.Client]): Custom httpx client for making non-tracing HTTP requests. If not provided, a default client will be created.
 160        debug (bool): Enable debug logging. Defaults to False. Can also be set via LANGFUSE_DEBUG environment variable.
 161        tracing_enabled (Optional[bool]): Enable or disable tracing. Defaults to True. Can also be set via LANGFUSE_TRACING_ENABLED environment variable.
 162        flush_at (Optional[int]): Number of spans to batch before sending to the API. Defaults to 512. Can also be set via LANGFUSE_FLUSH_AT environment variable.
 163        flush_interval (Optional[float]): Time in seconds between batch flushes. Defaults to 5 seconds. Can also be set via LANGFUSE_FLUSH_INTERVAL environment variable.
 164        environment (Optional[str]): Environment name for tracing. Default is 'default'. Can also be set via LANGFUSE_TRACING_ENVIRONMENT environment variable. Can be any lowercase alphanumeric string with hyphens and underscores that does not start with 'langfuse'.
 165        release (Optional[str]): Release version/hash of your application. Used for grouping analytics by release.
 166        media_upload_thread_count (Optional[int]): Number of background threads for handling media uploads. Defaults to 1. Can also be set via LANGFUSE_MEDIA_UPLOAD_THREAD_COUNT environment variable.
 167        sample_rate (Optional[float]): Sampling rate for traces (0.0 to 1.0). Defaults to 1.0 (100% of traces are sampled). Can also be set via LANGFUSE_SAMPLE_RATE environment variable.
 168        mask (Optional[MaskFunction]): Function to mask sensitive data in traces before sending to the API.
 169        blocked_instrumentation_scopes (Optional[List[str]]): List of instrumentation scope names to block from being exported to Langfuse. Spans from these scopes will be filtered out before being sent to the API. Useful for filtering out spans from specific libraries or frameworks. For exported spans, you can see the instrumentation scope name in the span metadata in Langfuse (`metadata.scope.name`)
 170        additional_headers (Optional[Dict[str, str]]): Additional headers to include in all API requests and OTLPSpanExporter requests. These headers will be merged with default headers. Note: If httpx_client is provided, additional_headers must be set directly on your custom httpx_client as well.
 171        tracer_provider(Optional[TracerProvider]): OpenTelemetry TracerProvider to use for Langfuse. This can be useful to set to have disconnected tracing between Langfuse and other OpenTelemetry-span emitting libraries. Note: To track active spans, the context is still shared between TracerProviders. This may lead to broken trace trees.
 172
 173    Example:
 174        ```python
 175        from langfuse.otel import Langfuse
 176
 177        # Initialize the client (reads from env vars if not provided)
 178        langfuse = Langfuse(
 179            public_key="your-public-key",
 180            secret_key="your-secret-key",
 181            host="https://cloud.langfuse.com",  # Optional, default shown
 182        )
 183
 184        # Create a trace span
 185        with langfuse.start_as_current_span(name="process-query") as span:
 186            # Your application code here
 187
 188            # Create a nested generation span for an LLM call
 189            with span.start_as_current_generation(
 190                name="generate-response",
 191                model="gpt-4",
 192                input={"query": "Tell me about AI"},
 193                model_parameters={"temperature": 0.7, "max_tokens": 500}
 194            ) as generation:
 195                # Generate response here
 196                response = "AI is a field of computer science..."
 197
 198                generation.update(
 199                    output=response,
 200                    usage_details={"prompt_tokens": 10, "completion_tokens": 50},
 201                    cost_details={"total_cost": 0.0023}
 202                )
 203
 204                # Score the generation (supports NUMERIC, BOOLEAN, CATEGORICAL)
 205                generation.score(name="relevance", value=0.95, data_type="NUMERIC")
 206        ```
 207    """
 208
 209    _resources: Optional[LangfuseResourceManager] = None
 210    _mask: Optional[MaskFunction] = None
 211    _otel_tracer: otel_trace_api.Tracer
 212
 213    def __init__(
 214        self,
 215        *,
 216        public_key: Optional[str] = None,
 217        secret_key: Optional[str] = None,
 218        base_url: Optional[str] = None,
 219        host: Optional[str] = None,
 220        timeout: Optional[int] = None,
 221        httpx_client: Optional[httpx.Client] = None,
 222        debug: bool = False,
 223        tracing_enabled: Optional[bool] = True,
 224        flush_at: Optional[int] = None,
 225        flush_interval: Optional[float] = None,
 226        environment: Optional[str] = None,
 227        release: Optional[str] = None,
 228        media_upload_thread_count: Optional[int] = None,
 229        sample_rate: Optional[float] = None,
 230        mask: Optional[MaskFunction] = None,
 231        blocked_instrumentation_scopes: Optional[List[str]] = None,
 232        additional_headers: Optional[Dict[str, str]] = None,
 233        tracer_provider: Optional[TracerProvider] = None,
 234    ):
 235        self._base_url = (
 236            base_url
 237            or os.environ.get(LANGFUSE_BASE_URL)
 238            or host
 239            or os.environ.get(LANGFUSE_HOST, "https://cloud.langfuse.com")
 240        )
 241        self._environment = environment or cast(
 242            str, os.environ.get(LANGFUSE_TRACING_ENVIRONMENT)
 243        )
 244        self._project_id: Optional[str] = None
 245        sample_rate = sample_rate or float(os.environ.get(LANGFUSE_SAMPLE_RATE, 1.0))
 246        if not 0.0 <= sample_rate <= 1.0:
 247            raise ValueError(
 248                f"Sample rate must be between 0.0 and 1.0, got {sample_rate}"
 249            )
 250
 251        timeout = timeout or int(os.environ.get(LANGFUSE_TIMEOUT, 5))
 252
 253        self._tracing_enabled = (
 254            tracing_enabled
 255            and os.environ.get(LANGFUSE_TRACING_ENABLED, "true").lower() != "false"
 256        )
 257        if not self._tracing_enabled:
 258            langfuse_logger.info(
 259                "Configuration: Langfuse tracing is explicitly disabled. No data will be sent to the Langfuse API."
 260            )
 261
 262        debug = (
 263            debug if debug else (os.getenv(LANGFUSE_DEBUG, "false").lower() == "true")
 264        )
 265        if debug:
 266            logging.basicConfig(
 267                format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
 268            )
 269            langfuse_logger.setLevel(logging.DEBUG)
 270
 271        public_key = public_key or os.environ.get(LANGFUSE_PUBLIC_KEY)
 272        if public_key is None:
 273            langfuse_logger.warning(
 274                "Authentication error: Langfuse client initialized without public_key. Client will be disabled. "
 275                "Provide a public_key parameter or set LANGFUSE_PUBLIC_KEY environment variable. "
 276            )
 277            self._otel_tracer = otel_trace_api.NoOpTracer()
 278            return
 279
 280        secret_key = secret_key or os.environ.get(LANGFUSE_SECRET_KEY)
 281        if secret_key is None:
 282            langfuse_logger.warning(
 283                "Authentication error: Langfuse client initialized without secret_key. Client will be disabled. "
 284                "Provide a secret_key parameter or set LANGFUSE_SECRET_KEY environment variable. "
 285            )
 286            self._otel_tracer = otel_trace_api.NoOpTracer()
 287            return
 288
 289        if os.environ.get("OTEL_SDK_DISABLED", "false").lower() == "true":
 290            langfuse_logger.warning(
 291                "OTEL_SDK_DISABLED is set. Langfuse tracing will be disabled and no traces will appear in the UI."
 292            )
 293
 294        # Initialize api and tracer if requirements are met
 295        self._resources = LangfuseResourceManager(
 296            public_key=public_key,
 297            secret_key=secret_key,
 298            base_url=self._base_url,
 299            timeout=timeout,
 300            environment=self._environment,
 301            release=release,
 302            flush_at=flush_at,
 303            flush_interval=flush_interval,
 304            httpx_client=httpx_client,
 305            media_upload_thread_count=media_upload_thread_count,
 306            sample_rate=sample_rate,
 307            mask=mask,
 308            tracing_enabled=self._tracing_enabled,
 309            blocked_instrumentation_scopes=blocked_instrumentation_scopes,
 310            additional_headers=additional_headers,
 311            tracer_provider=tracer_provider,
 312        )
 313        self._mask = self._resources.mask
 314
 315        self._otel_tracer = (
 316            self._resources.tracer
 317            if self._tracing_enabled and self._resources.tracer is not None
 318            else otel_trace_api.NoOpTracer()
 319        )
 320        self.api = self._resources.api
 321        self.async_api = self._resources.async_api
 322
 323    def start_span(
 324        self,
 325        *,
 326        trace_context: Optional[TraceContext] = None,
 327        name: str,
 328        input: Optional[Any] = None,
 329        output: Optional[Any] = None,
 330        metadata: Optional[Any] = None,
 331        version: Optional[str] = None,
 332        level: Optional[SpanLevel] = None,
 333        status_message: Optional[str] = None,
 334    ) -> LangfuseSpan:
 335        """Create a new span for tracing a unit of work.
 336
 337        This method creates a new span but does not set it as the current span in the
 338        context. To create and use a span within a context, use start_as_current_span().
 339
 340        The created span will be the child of the current span in the context.
 341
 342        Args:
 343            trace_context: Optional context for connecting to an existing trace
 344            name: Name of the span (e.g., function or operation name)
 345            input: Input data for the operation (can be any JSON-serializable object)
 346            output: Output data from the operation (can be any JSON-serializable object)
 347            metadata: Additional metadata to associate with the span
 348            version: Version identifier for the code or component
 349            level: Importance level of the span (info, warning, error)
 350            status_message: Optional status message for the span
 351
 352        Returns:
 353            A LangfuseSpan object that must be ended with .end() when the operation completes
 354
 355        Example:
 356            ```python
 357            span = langfuse.start_span(name="process-data")
 358            try:
 359                # Do work
 360                span.update(output="result")
 361            finally:
 362                span.end()
 363            ```
 364        """
 365        return self.start_observation(
 366            trace_context=trace_context,
 367            name=name,
 368            as_type="span",
 369            input=input,
 370            output=output,
 371            metadata=metadata,
 372            version=version,
 373            level=level,
 374            status_message=status_message,
 375        )
 376
 377    def start_as_current_span(
 378        self,
 379        *,
 380        trace_context: Optional[TraceContext] = None,
 381        name: str,
 382        input: Optional[Any] = None,
 383        output: Optional[Any] = None,
 384        metadata: Optional[Any] = None,
 385        version: Optional[str] = None,
 386        level: Optional[SpanLevel] = None,
 387        status_message: Optional[str] = None,
 388        end_on_exit: Optional[bool] = None,
 389    ) -> _AgnosticContextManager[LangfuseSpan]:
 390        """Create a new span and set it as the current span in a context manager.
 391
 392        This method creates a new span and sets it as the current span within a context
 393        manager. Use this method with a 'with' statement to automatically handle span
 394        lifecycle within a code block.
 395
 396        The created span will be the child of the current span in the context.
 397
 398        Args:
 399            trace_context: Optional context for connecting to an existing trace
 400            name: Name of the span (e.g., function or operation name)
 401            input: Input data for the operation (can be any JSON-serializable object)
 402            output: Output data from the operation (can be any JSON-serializable object)
 403            metadata: Additional metadata to associate with the span
 404            version: Version identifier for the code or component
 405            level: Importance level of the span (info, warning, error)
 406            status_message: Optional status message for the span
 407            end_on_exit (default: True): Whether to end the span automatically when leaving the context manager. If False, the span must be manually ended to avoid memory leaks.
 408
 409        Returns:
 410            A context manager that yields a LangfuseSpan
 411
 412        Example:
 413            ```python
 414            with langfuse.start_as_current_span(name="process-query") as span:
 415                # Do work
 416                result = process_data()
 417                span.update(output=result)
 418
 419                # Create a child span automatically
 420                with span.start_as_current_span(name="sub-operation") as child_span:
 421                    # Do sub-operation work
 422                    child_span.update(output="sub-result")
 423            ```
 424        """
 425        return self.start_as_current_observation(
 426            trace_context=trace_context,
 427            name=name,
 428            as_type="span",
 429            input=input,
 430            output=output,
 431            metadata=metadata,
 432            version=version,
 433            level=level,
 434            status_message=status_message,
 435            end_on_exit=end_on_exit,
 436        )
 437
 438    @overload
 439    def start_observation(
 440        self,
 441        *,
 442        trace_context: Optional[TraceContext] = None,
 443        name: str,
 444        as_type: Literal["generation"],
 445        input: Optional[Any] = None,
 446        output: Optional[Any] = None,
 447        metadata: Optional[Any] = None,
 448        version: Optional[str] = None,
 449        level: Optional[SpanLevel] = None,
 450        status_message: Optional[str] = None,
 451        completion_start_time: Optional[datetime] = None,
 452        model: Optional[str] = None,
 453        model_parameters: Optional[Dict[str, MapValue]] = None,
 454        usage_details: Optional[Dict[str, int]] = None,
 455        cost_details: Optional[Dict[str, float]] = None,
 456        prompt: Optional[PromptClient] = None,
 457    ) -> LangfuseGeneration: ...
 458
 459    @overload
 460    def start_observation(
 461        self,
 462        *,
 463        trace_context: Optional[TraceContext] = None,
 464        name: str,
 465        as_type: Literal["span"] = "span",
 466        input: Optional[Any] = None,
 467        output: Optional[Any] = None,
 468        metadata: Optional[Any] = None,
 469        version: Optional[str] = None,
 470        level: Optional[SpanLevel] = None,
 471        status_message: Optional[str] = None,
 472    ) -> LangfuseSpan: ...
 473
 474    @overload
 475    def start_observation(
 476        self,
 477        *,
 478        trace_context: Optional[TraceContext] = None,
 479        name: str,
 480        as_type: Literal["agent"],
 481        input: Optional[Any] = None,
 482        output: Optional[Any] = None,
 483        metadata: Optional[Any] = None,
 484        version: Optional[str] = None,
 485        level: Optional[SpanLevel] = None,
 486        status_message: Optional[str] = None,
 487    ) -> LangfuseAgent: ...
 488
 489    @overload
 490    def start_observation(
 491        self,
 492        *,
 493        trace_context: Optional[TraceContext] = None,
 494        name: str,
 495        as_type: Literal["tool"],
 496        input: Optional[Any] = None,
 497        output: Optional[Any] = None,
 498        metadata: Optional[Any] = None,
 499        version: Optional[str] = None,
 500        level: Optional[SpanLevel] = None,
 501        status_message: Optional[str] = None,
 502    ) -> LangfuseTool: ...
 503
 504    @overload
 505    def start_observation(
 506        self,
 507        *,
 508        trace_context: Optional[TraceContext] = None,
 509        name: str,
 510        as_type: Literal["chain"],
 511        input: Optional[Any] = None,
 512        output: Optional[Any] = None,
 513        metadata: Optional[Any] = None,
 514        version: Optional[str] = None,
 515        level: Optional[SpanLevel] = None,
 516        status_message: Optional[str] = None,
 517    ) -> LangfuseChain: ...
 518
 519    @overload
 520    def start_observation(
 521        self,
 522        *,
 523        trace_context: Optional[TraceContext] = None,
 524        name: str,
 525        as_type: Literal["retriever"],
 526        input: Optional[Any] = None,
 527        output: Optional[Any] = None,
 528        metadata: Optional[Any] = None,
 529        version: Optional[str] = None,
 530        level: Optional[SpanLevel] = None,
 531        status_message: Optional[str] = None,
 532    ) -> LangfuseRetriever: ...
 533
 534    @overload
 535    def start_observation(
 536        self,
 537        *,
 538        trace_context: Optional[TraceContext] = None,
 539        name: str,
 540        as_type: Literal["evaluator"],
 541        input: Optional[Any] = None,
 542        output: Optional[Any] = None,
 543        metadata: Optional[Any] = None,
 544        version: Optional[str] = None,
 545        level: Optional[SpanLevel] = None,
 546        status_message: Optional[str] = None,
 547    ) -> LangfuseEvaluator: ...
 548
 549    @overload
 550    def start_observation(
 551        self,
 552        *,
 553        trace_context: Optional[TraceContext] = None,
 554        name: str,
 555        as_type: Literal["embedding"],
 556        input: Optional[Any] = None,
 557        output: Optional[Any] = None,
 558        metadata: Optional[Any] = None,
 559        version: Optional[str] = None,
 560        level: Optional[SpanLevel] = None,
 561        status_message: Optional[str] = None,
 562        completion_start_time: Optional[datetime] = None,
 563        model: Optional[str] = None,
 564        model_parameters: Optional[Dict[str, MapValue]] = None,
 565        usage_details: Optional[Dict[str, int]] = None,
 566        cost_details: Optional[Dict[str, float]] = None,
 567        prompt: Optional[PromptClient] = None,
 568    ) -> LangfuseEmbedding: ...
 569
 570    @overload
 571    def start_observation(
 572        self,
 573        *,
 574        trace_context: Optional[TraceContext] = None,
 575        name: str,
 576        as_type: Literal["guardrail"],
 577        input: Optional[Any] = None,
 578        output: Optional[Any] = None,
 579        metadata: Optional[Any] = None,
 580        version: Optional[str] = None,
 581        level: Optional[SpanLevel] = None,
 582        status_message: Optional[str] = None,
 583    ) -> LangfuseGuardrail: ...
 584
 585    def start_observation(
 586        self,
 587        *,
 588        trace_context: Optional[TraceContext] = None,
 589        name: str,
 590        as_type: ObservationTypeLiteralNoEvent = "span",
 591        input: Optional[Any] = None,
 592        output: Optional[Any] = None,
 593        metadata: Optional[Any] = None,
 594        version: Optional[str] = None,
 595        level: Optional[SpanLevel] = None,
 596        status_message: Optional[str] = None,
 597        completion_start_time: Optional[datetime] = None,
 598        model: Optional[str] = None,
 599        model_parameters: Optional[Dict[str, MapValue]] = None,
 600        usage_details: Optional[Dict[str, int]] = None,
 601        cost_details: Optional[Dict[str, float]] = None,
 602        prompt: Optional[PromptClient] = None,
 603    ) -> Union[
 604        LangfuseSpan,
 605        LangfuseGeneration,
 606        LangfuseAgent,
 607        LangfuseTool,
 608        LangfuseChain,
 609        LangfuseRetriever,
 610        LangfuseEvaluator,
 611        LangfuseEmbedding,
 612        LangfuseGuardrail,
 613    ]:
 614        """Create a new observation of the specified type.
 615
 616        This method creates a new observation but does not set it as the current span in the
 617        context. To create and use an observation within a context, use start_as_current_observation().
 618
 619        Args:
 620            trace_context: Optional context for connecting to an existing trace
 621            name: Name of the observation
 622            as_type: Type of observation to create (defaults to "span")
 623            input: Input data for the operation
 624            output: Output data from the operation
 625            metadata: Additional metadata to associate with the observation
 626            version: Version identifier for the code or component
 627            level: Importance level of the observation
 628            status_message: Optional status message for the observation
 629            completion_start_time: When the model started generating (for generation types)
 630            model: Name/identifier of the AI model used (for generation types)
 631            model_parameters: Parameters used for the model (for generation types)
 632            usage_details: Token usage information (for generation types)
 633            cost_details: Cost information (for generation types)
 634            prompt: Associated prompt template (for generation types)
 635
 636        Returns:
 637            An observation object of the appropriate type that must be ended with .end()
 638        """
 639        if trace_context:
 640            trace_id = trace_context.get("trace_id", None)
 641            parent_span_id = trace_context.get("parent_span_id", None)
 642
 643            if trace_id:
 644                remote_parent_span = self._create_remote_parent_span(
 645                    trace_id=trace_id, parent_span_id=parent_span_id
 646                )
 647
 648                with otel_trace_api.use_span(
 649                    cast(otel_trace_api.Span, remote_parent_span)
 650                ):
 651                    otel_span = self._otel_tracer.start_span(name=name)
 652                    otel_span.set_attribute(LangfuseOtelSpanAttributes.AS_ROOT, True)
 653
 654                    return self._create_observation_from_otel_span(
 655                        otel_span=otel_span,
 656                        as_type=as_type,
 657                        input=input,
 658                        output=output,
 659                        metadata=metadata,
 660                        version=version,
 661                        level=level,
 662                        status_message=status_message,
 663                        completion_start_time=completion_start_time,
 664                        model=model,
 665                        model_parameters=model_parameters,
 666                        usage_details=usage_details,
 667                        cost_details=cost_details,
 668                        prompt=prompt,
 669                    )
 670
 671        otel_span = self._otel_tracer.start_span(name=name)
 672
 673        return self._create_observation_from_otel_span(
 674            otel_span=otel_span,
 675            as_type=as_type,
 676            input=input,
 677            output=output,
 678            metadata=metadata,
 679            version=version,
 680            level=level,
 681            status_message=status_message,
 682            completion_start_time=completion_start_time,
 683            model=model,
 684            model_parameters=model_parameters,
 685            usage_details=usage_details,
 686            cost_details=cost_details,
 687            prompt=prompt,
 688        )
 689
 690    def _create_observation_from_otel_span(
 691        self,
 692        *,
 693        otel_span: otel_trace_api.Span,
 694        as_type: ObservationTypeLiteralNoEvent,
 695        input: Optional[Any] = None,
 696        output: Optional[Any] = None,
 697        metadata: Optional[Any] = None,
 698        version: Optional[str] = None,
 699        level: Optional[SpanLevel] = None,
 700        status_message: Optional[str] = None,
 701        completion_start_time: Optional[datetime] = None,
 702        model: Optional[str] = None,
 703        model_parameters: Optional[Dict[str, MapValue]] = None,
 704        usage_details: Optional[Dict[str, int]] = None,
 705        cost_details: Optional[Dict[str, float]] = None,
 706        prompt: Optional[PromptClient] = None,
 707    ) -> Union[
 708        LangfuseSpan,
 709        LangfuseGeneration,
 710        LangfuseAgent,
 711        LangfuseTool,
 712        LangfuseChain,
 713        LangfuseRetriever,
 714        LangfuseEvaluator,
 715        LangfuseEmbedding,
 716        LangfuseGuardrail,
 717    ]:
 718        """Create the appropriate observation type from an OTEL span."""
 719        if as_type in get_observation_types_list(ObservationTypeGenerationLike):
 720            observation_class = self._get_span_class(as_type)
 721            # Type ignore to prevent overloads of internal _get_span_class function,
 722            # issue is that LangfuseEvent could be returned and that classes have diff. args
 723            return observation_class(  # type: ignore[return-value,call-arg]
 724                otel_span=otel_span,
 725                langfuse_client=self,
 726                environment=self._environment,
 727                input=input,
 728                output=output,
 729                metadata=metadata,
 730                version=version,
 731                level=level,
 732                status_message=status_message,
 733                completion_start_time=completion_start_time,
 734                model=model,
 735                model_parameters=model_parameters,
 736                usage_details=usage_details,
 737                cost_details=cost_details,
 738                prompt=prompt,
 739            )
 740        else:
 741            # For other types (e.g. span, guardrail), create appropriate class without generation properties
 742            observation_class = self._get_span_class(as_type)
 743            # Type ignore to prevent overloads of internal _get_span_class function,
 744            # issue is that LangfuseEvent could be returned and that classes have diff. args
 745            return observation_class(  # type: ignore[return-value,call-arg]
 746                otel_span=otel_span,
 747                langfuse_client=self,
 748                environment=self._environment,
 749                input=input,
 750                output=output,
 751                metadata=metadata,
 752                version=version,
 753                level=level,
 754                status_message=status_message,
 755            )
 756            # span._observation_type = as_type
 757            # span._otel_span.set_attribute("langfuse.observation.type", as_type)
 758            # return span
 759
 760    def start_generation(
 761        self,
 762        *,
 763        trace_context: Optional[TraceContext] = None,
 764        name: str,
 765        input: Optional[Any] = None,
 766        output: Optional[Any] = None,
 767        metadata: Optional[Any] = None,
 768        version: Optional[str] = None,
 769        level: Optional[SpanLevel] = None,
 770        status_message: Optional[str] = None,
 771        completion_start_time: Optional[datetime] = None,
 772        model: Optional[str] = None,
 773        model_parameters: Optional[Dict[str, MapValue]] = None,
 774        usage_details: Optional[Dict[str, int]] = None,
 775        cost_details: Optional[Dict[str, float]] = None,
 776        prompt: Optional[PromptClient] = None,
 777    ) -> LangfuseGeneration:
 778        """Create a new generation span for model generations.
 779
 780        DEPRECATED: This method is deprecated and will be removed in a future version.
 781        Use start_observation(as_type='generation') instead.
 782
 783        This method creates a specialized span for tracking model generations.
 784        It includes additional fields specific to model generations such as model name,
 785        token usage, and cost details.
 786
 787        The created generation span will be the child of the current span in the context.
 788
 789        Args:
 790            trace_context: Optional context for connecting to an existing trace
 791            name: Name of the generation operation
 792            input: Input data for the model (e.g., prompts)
 793            output: Output from the model (e.g., completions)
 794            metadata: Additional metadata to associate with the generation
 795            version: Version identifier for the model or component
 796            level: Importance level of the generation (info, warning, error)
 797            status_message: Optional status message for the generation
 798            completion_start_time: When the model started generating the response
 799            model: Name/identifier of the AI model used (e.g., "gpt-4")
 800            model_parameters: Parameters used for the model (e.g., temperature, max_tokens)
 801            usage_details: Token usage information (e.g., prompt_tokens, completion_tokens)
 802            cost_details: Cost information for the model call
 803            prompt: Associated prompt template from Langfuse prompt management
 804
 805        Returns:
 806            A LangfuseGeneration object that must be ended with .end() when complete
 807
 808        Example:
 809            ```python
 810            generation = langfuse.start_generation(
 811                name="answer-generation",
 812                model="gpt-4",
 813                input={"prompt": "Explain quantum computing"},
 814                model_parameters={"temperature": 0.7}
 815            )
 816            try:
 817                # Call model API
 818                response = llm.generate(...)
 819
 820                generation.update(
 821                    output=response.text,
 822                    usage_details={
 823                        "prompt_tokens": response.usage.prompt_tokens,
 824                        "completion_tokens": response.usage.completion_tokens
 825                    }
 826                )
 827            finally:
 828                generation.end()
 829            ```
 830        """
 831        warnings.warn(
 832            "start_generation is deprecated and will be removed in a future version. "
 833            "Use start_observation(as_type='generation') instead.",
 834            DeprecationWarning,
 835            stacklevel=2,
 836        )
 837        return self.start_observation(
 838            trace_context=trace_context,
 839            name=name,
 840            as_type="generation",
 841            input=input,
 842            output=output,
 843            metadata=metadata,
 844            version=version,
 845            level=level,
 846            status_message=status_message,
 847            completion_start_time=completion_start_time,
 848            model=model,
 849            model_parameters=model_parameters,
 850            usage_details=usage_details,
 851            cost_details=cost_details,
 852            prompt=prompt,
 853        )
 854
 855    def start_as_current_generation(
 856        self,
 857        *,
 858        trace_context: Optional[TraceContext] = None,
 859        name: str,
 860        input: Optional[Any] = None,
 861        output: Optional[Any] = None,
 862        metadata: Optional[Any] = None,
 863        version: Optional[str] = None,
 864        level: Optional[SpanLevel] = None,
 865        status_message: Optional[str] = None,
 866        completion_start_time: Optional[datetime] = None,
 867        model: Optional[str] = None,
 868        model_parameters: Optional[Dict[str, MapValue]] = None,
 869        usage_details: Optional[Dict[str, int]] = None,
 870        cost_details: Optional[Dict[str, float]] = None,
 871        prompt: Optional[PromptClient] = None,
 872        end_on_exit: Optional[bool] = None,
 873    ) -> _AgnosticContextManager[LangfuseGeneration]:
 874        """Create a new generation span and set it as the current span in a context manager.
 875
 876        DEPRECATED: This method is deprecated and will be removed in a future version.
 877        Use start_as_current_observation(as_type='generation') instead.
 878
 879        This method creates a specialized span for model generations and sets it as the
 880        current span within a context manager. Use this method with a 'with' statement to
 881        automatically handle the generation span lifecycle within a code block.
 882
 883        The created generation span will be the child of the current span in the context.
 884
 885        Args:
 886            trace_context: Optional context for connecting to an existing trace
 887            name: Name of the generation operation
 888            input: Input data for the model (e.g., prompts)
 889            output: Output from the model (e.g., completions)
 890            metadata: Additional metadata to associate with the generation
 891            version: Version identifier for the model or component
 892            level: Importance level of the generation (info, warning, error)
 893            status_message: Optional status message for the generation
 894            completion_start_time: When the model started generating the response
 895            model: Name/identifier of the AI model used (e.g., "gpt-4")
 896            model_parameters: Parameters used for the model (e.g., temperature, max_tokens)
 897            usage_details: Token usage information (e.g., prompt_tokens, completion_tokens)
 898            cost_details: Cost information for the model call
 899            prompt: Associated prompt template from Langfuse prompt management
 900            end_on_exit (default: True): Whether to end the span automatically when leaving the context manager. If False, the span must be manually ended to avoid memory leaks.
 901
 902        Returns:
 903            A context manager that yields a LangfuseGeneration
 904
 905        Example:
 906            ```python
 907            with langfuse.start_as_current_generation(
 908                name="answer-generation",
 909                model="gpt-4",
 910                input={"prompt": "Explain quantum computing"}
 911            ) as generation:
 912                # Call model API
 913                response = llm.generate(...)
 914
 915                # Update with results
 916                generation.update(
 917                    output=response.text,
 918                    usage_details={
 919                        "prompt_tokens": response.usage.prompt_tokens,
 920                        "completion_tokens": response.usage.completion_tokens
 921                    }
 922                )
 923            ```
 924        """
 925        warnings.warn(
 926            "start_as_current_generation is deprecated and will be removed in a future version. "
 927            "Use start_as_current_observation(as_type='generation') instead.",
 928            DeprecationWarning,
 929            stacklevel=2,
 930        )
 931        return self.start_as_current_observation(
 932            trace_context=trace_context,
 933            name=name,
 934            as_type="generation",
 935            input=input,
 936            output=output,
 937            metadata=metadata,
 938            version=version,
 939            level=level,
 940            status_message=status_message,
 941            completion_start_time=completion_start_time,
 942            model=model,
 943            model_parameters=model_parameters,
 944            usage_details=usage_details,
 945            cost_details=cost_details,
 946            prompt=prompt,
 947            end_on_exit=end_on_exit,
 948        )
 949
 950    @overload
 951    def start_as_current_observation(
 952        self,
 953        *,
 954        trace_context: Optional[TraceContext] = None,
 955        name: str,
 956        as_type: Literal["generation"],
 957        input: Optional[Any] = None,
 958        output: Optional[Any] = None,
 959        metadata: Optional[Any] = None,
 960        version: Optional[str] = None,
 961        level: Optional[SpanLevel] = None,
 962        status_message: Optional[str] = None,
 963        completion_start_time: Optional[datetime] = None,
 964        model: Optional[str] = None,
 965        model_parameters: Optional[Dict[str, MapValue]] = None,
 966        usage_details: Optional[Dict[str, int]] = None,
 967        cost_details: Optional[Dict[str, float]] = None,
 968        prompt: Optional[PromptClient] = None,
 969        end_on_exit: Optional[bool] = None,
 970    ) -> _AgnosticContextManager[LangfuseGeneration]: ...
 971
 972    @overload
 973    def start_as_current_observation(
 974        self,
 975        *,
 976        trace_context: Optional[TraceContext] = None,
 977        name: str,
 978        as_type: Literal["span"] = "span",
 979        input: Optional[Any] = None,
 980        output: Optional[Any] = None,
 981        metadata: Optional[Any] = None,
 982        version: Optional[str] = None,
 983        level: Optional[SpanLevel] = None,
 984        status_message: Optional[str] = None,
 985        end_on_exit: Optional[bool] = None,
 986    ) -> _AgnosticContextManager[LangfuseSpan]: ...
 987
 988    @overload
 989    def start_as_current_observation(
 990        self,
 991        *,
 992        trace_context: Optional[TraceContext] = None,
 993        name: str,
 994        as_type: Literal["agent"],
 995        input: Optional[Any] = None,
 996        output: Optional[Any] = None,
 997        metadata: Optional[Any] = None,
 998        version: Optional[str] = None,
 999        level: Optional[SpanLevel] = None,
1000        status_message: Optional[str] = None,
1001        end_on_exit: Optional[bool] = None,
1002    ) -> _AgnosticContextManager[LangfuseAgent]: ...
1003
1004    @overload
1005    def start_as_current_observation(
1006        self,
1007        *,
1008        trace_context: Optional[TraceContext] = None,
1009        name: str,
1010        as_type: Literal["tool"],
1011        input: Optional[Any] = None,
1012        output: Optional[Any] = None,
1013        metadata: Optional[Any] = None,
1014        version: Optional[str] = None,
1015        level: Optional[SpanLevel] = None,
1016        status_message: Optional[str] = None,
1017        end_on_exit: Optional[bool] = None,
1018    ) -> _AgnosticContextManager[LangfuseTool]: ...
1019
1020    @overload
1021    def start_as_current_observation(
1022        self,
1023        *,
1024        trace_context: Optional[TraceContext] = None,
1025        name: str,
1026        as_type: Literal["chain"],
1027        input: Optional[Any] = None,
1028        output: Optional[Any] = None,
1029        metadata: Optional[Any] = None,
1030        version: Optional[str] = None,
1031        level: Optional[SpanLevel] = None,
1032        status_message: Optional[str] = None,
1033        end_on_exit: Optional[bool] = None,
1034    ) -> _AgnosticContextManager[LangfuseChain]: ...
1035
1036    @overload
1037    def start_as_current_observation(
1038        self,
1039        *,
1040        trace_context: Optional[TraceContext] = None,
1041        name: str,
1042        as_type: Literal["retriever"],
1043        input: Optional[Any] = None,
1044        output: Optional[Any] = None,
1045        metadata: Optional[Any] = None,
1046        version: Optional[str] = None,
1047        level: Optional[SpanLevel] = None,
1048        status_message: Optional[str] = None,
1049        end_on_exit: Optional[bool] = None,
1050    ) -> _AgnosticContextManager[LangfuseRetriever]: ...
1051
1052    @overload
1053    def start_as_current_observation(
1054        self,
1055        *,
1056        trace_context: Optional[TraceContext] = None,
1057        name: str,
1058        as_type: Literal["evaluator"],
1059        input: Optional[Any] = None,
1060        output: Optional[Any] = None,
1061        metadata: Optional[Any] = None,
1062        version: Optional[str] = None,
1063        level: Optional[SpanLevel] = None,
1064        status_message: Optional[str] = None,
1065        end_on_exit: Optional[bool] = None,
1066    ) -> _AgnosticContextManager[LangfuseEvaluator]: ...
1067
1068    @overload
1069    def start_as_current_observation(
1070        self,
1071        *,
1072        trace_context: Optional[TraceContext] = None,
1073        name: str,
1074        as_type: Literal["embedding"],
1075        input: Optional[Any] = None,
1076        output: Optional[Any] = None,
1077        metadata: Optional[Any] = None,
1078        version: Optional[str] = None,
1079        level: Optional[SpanLevel] = None,
1080        status_message: Optional[str] = None,
1081        completion_start_time: Optional[datetime] = None,
1082        model: Optional[str] = None,
1083        model_parameters: Optional[Dict[str, MapValue]] = None,
1084        usage_details: Optional[Dict[str, int]] = None,
1085        cost_details: Optional[Dict[str, float]] = None,
1086        prompt: Optional[PromptClient] = None,
1087        end_on_exit: Optional[bool] = None,
1088    ) -> _AgnosticContextManager[LangfuseEmbedding]: ...
1089
1090    @overload
1091    def start_as_current_observation(
1092        self,
1093        *,
1094        trace_context: Optional[TraceContext] = None,
1095        name: str,
1096        as_type: Literal["guardrail"],
1097        input: Optional[Any] = None,
1098        output: Optional[Any] = None,
1099        metadata: Optional[Any] = None,
1100        version: Optional[str] = None,
1101        level: Optional[SpanLevel] = None,
1102        status_message: Optional[str] = None,
1103        end_on_exit: Optional[bool] = None,
1104    ) -> _AgnosticContextManager[LangfuseGuardrail]: ...
1105
1106    def start_as_current_observation(
1107        self,
1108        *,
1109        trace_context: Optional[TraceContext] = None,
1110        name: str,
1111        as_type: ObservationTypeLiteralNoEvent = "span",
1112        input: Optional[Any] = None,
1113        output: Optional[Any] = None,
1114        metadata: Optional[Any] = None,
1115        version: Optional[str] = None,
1116        level: Optional[SpanLevel] = None,
1117        status_message: Optional[str] = None,
1118        completion_start_time: Optional[datetime] = None,
1119        model: Optional[str] = None,
1120        model_parameters: Optional[Dict[str, MapValue]] = None,
1121        usage_details: Optional[Dict[str, int]] = None,
1122        cost_details: Optional[Dict[str, float]] = None,
1123        prompt: Optional[PromptClient] = None,
1124        end_on_exit: Optional[bool] = None,
1125    ) -> Union[
1126        _AgnosticContextManager[LangfuseGeneration],
1127        _AgnosticContextManager[LangfuseSpan],
1128        _AgnosticContextManager[LangfuseAgent],
1129        _AgnosticContextManager[LangfuseTool],
1130        _AgnosticContextManager[LangfuseChain],
1131        _AgnosticContextManager[LangfuseRetriever],
1132        _AgnosticContextManager[LangfuseEvaluator],
1133        _AgnosticContextManager[LangfuseEmbedding],
1134        _AgnosticContextManager[LangfuseGuardrail],
1135    ]:
1136        """Create a new observation and set it as the current span in a context manager.
1137
1138        This method creates a new observation of the specified type and sets it as the
1139        current span within a context manager. Use this method with a 'with' statement to
1140        automatically handle the observation lifecycle within a code block.
1141
1142        The created observation will be the child of the current span in the context.
1143
1144        Args:
1145            trace_context: Optional context for connecting to an existing trace
1146            name: Name of the observation (e.g., function or operation name)
1147            as_type: Type of observation to create (defaults to "span")
1148            input: Input data for the operation (can be any JSON-serializable object)
1149            output: Output data from the operation (can be any JSON-serializable object)
1150            metadata: Additional metadata to associate with the observation
1151            version: Version identifier for the code or component
1152            level: Importance level of the observation (info, warning, error)
1153            status_message: Optional status message for the observation
1154            end_on_exit (default: True): Whether to end the span automatically when leaving the context manager. If False, the span must be manually ended to avoid memory leaks.
1155
1156            The following parameters are available when as_type is: "generation" or "embedding".
1157            completion_start_time: When the model started generating the response
1158            model: Name/identifier of the AI model used (e.g., "gpt-4")
1159            model_parameters: Parameters used for the model (e.g., temperature, max_tokens)
1160            usage_details: Token usage information (e.g., prompt_tokens, completion_tokens)
1161            cost_details: Cost information for the model call
1162            prompt: Associated prompt template from Langfuse prompt management
1163
1164        Returns:
1165            A context manager that yields the appropriate observation type based on as_type
1166
1167        Example:
1168            ```python
1169            # Create a span
1170            with langfuse.start_as_current_observation(name="process-query", as_type="span") as span:
1171                # Do work
1172                result = process_data()
1173                span.update(output=result)
1174
1175                # Create a child span automatically
1176                with span.start_as_current_span(name="sub-operation") as child_span:
1177                    # Do sub-operation work
1178                    child_span.update(output="sub-result")
1179
1180            # Create a tool observation
1181            with langfuse.start_as_current_observation(name="web-search", as_type="tool") as tool:
1182                # Do tool work
1183                results = search_web(query)
1184                tool.update(output=results)
1185
1186            # Create a generation observation
1187            with langfuse.start_as_current_observation(
1188                name="answer-generation",
1189                as_type="generation",
1190                model="gpt-4"
1191            ) as generation:
1192                # Generate answer
1193                response = llm.generate(...)
1194                generation.update(output=response)
1195            ```
1196        """
1197        if as_type in get_observation_types_list(ObservationTypeGenerationLike):
1198            if trace_context:
1199                trace_id = trace_context.get("trace_id", None)
1200                parent_span_id = trace_context.get("parent_span_id", None)
1201
1202                if trace_id:
1203                    remote_parent_span = self._create_remote_parent_span(
1204                        trace_id=trace_id, parent_span_id=parent_span_id
1205                    )
1206
1207                    return cast(
1208                        Union[
1209                            _AgnosticContextManager[LangfuseGeneration],
1210                            _AgnosticContextManager[LangfuseEmbedding],
1211                        ],
1212                        self._create_span_with_parent_context(
1213                            as_type=as_type,
1214                            name=name,
1215                            remote_parent_span=remote_parent_span,
1216                            parent=None,
1217                            end_on_exit=end_on_exit,
1218                            input=input,
1219                            output=output,
1220                            metadata=metadata,
1221                            version=version,
1222                            level=level,
1223                            status_message=status_message,
1224                            completion_start_time=completion_start_time,
1225                            model=model,
1226                            model_parameters=model_parameters,
1227                            usage_details=usage_details,
1228                            cost_details=cost_details,
1229                            prompt=prompt,
1230                        ),
1231                    )
1232
1233            return cast(
1234                Union[
1235                    _AgnosticContextManager[LangfuseGeneration],
1236                    _AgnosticContextManager[LangfuseEmbedding],
1237                ],
1238                self._start_as_current_otel_span_with_processed_media(
1239                    as_type=as_type,
1240                    name=name,
1241                    end_on_exit=end_on_exit,
1242                    input=input,
1243                    output=output,
1244                    metadata=metadata,
1245                    version=version,
1246                    level=level,
1247                    status_message=status_message,
1248                    completion_start_time=completion_start_time,
1249                    model=model,
1250                    model_parameters=model_parameters,
1251                    usage_details=usage_details,
1252                    cost_details=cost_details,
1253                    prompt=prompt,
1254                ),
1255            )
1256
1257        if as_type in get_observation_types_list(ObservationTypeSpanLike):
1258            if trace_context:
1259                trace_id = trace_context.get("trace_id", None)
1260                parent_span_id = trace_context.get("parent_span_id", None)
1261
1262                if trace_id:
1263                    remote_parent_span = self._create_remote_parent_span(
1264                        trace_id=trace_id, parent_span_id=parent_span_id
1265                    )
1266
1267                    return cast(
1268                        Union[
1269                            _AgnosticContextManager[LangfuseSpan],
1270                            _AgnosticContextManager[LangfuseAgent],
1271                            _AgnosticContextManager[LangfuseTool],
1272                            _AgnosticContextManager[LangfuseChain],
1273                            _AgnosticContextManager[LangfuseRetriever],
1274                            _AgnosticContextManager[LangfuseEvaluator],
1275                            _AgnosticContextManager[LangfuseGuardrail],
1276                        ],
1277                        self._create_span_with_parent_context(
1278                            as_type=as_type,
1279                            name=name,
1280                            remote_parent_span=remote_parent_span,
1281                            parent=None,
1282                            end_on_exit=end_on_exit,
1283                            input=input,
1284                            output=output,
1285                            metadata=metadata,
1286                            version=version,
1287                            level=level,
1288                            status_message=status_message,
1289                        ),
1290                    )
1291
1292            return cast(
1293                Union[
1294                    _AgnosticContextManager[LangfuseSpan],
1295                    _AgnosticContextManager[LangfuseAgent],
1296                    _AgnosticContextManager[LangfuseTool],
1297                    _AgnosticContextManager[LangfuseChain],
1298                    _AgnosticContextManager[LangfuseRetriever],
1299                    _AgnosticContextManager[LangfuseEvaluator],
1300                    _AgnosticContextManager[LangfuseGuardrail],
1301                ],
1302                self._start_as_current_otel_span_with_processed_media(
1303                    as_type=as_type,
1304                    name=name,
1305                    end_on_exit=end_on_exit,
1306                    input=input,
1307                    output=output,
1308                    metadata=metadata,
1309                    version=version,
1310                    level=level,
1311                    status_message=status_message,
1312                ),
1313            )
1314
1315        # This should never be reached since all valid types are handled above
1316        langfuse_logger.warning(
1317            f"Unknown observation type: {as_type}, falling back to span"
1318        )
1319        return self._start_as_current_otel_span_with_processed_media(
1320            as_type="span",
1321            name=name,
1322            end_on_exit=end_on_exit,
1323            input=input,
1324            output=output,
1325            metadata=metadata,
1326            version=version,
1327            level=level,
1328            status_message=status_message,
1329        )
1330
1331    def _get_span_class(
1332        self,
1333        as_type: ObservationTypeLiteral,
1334    ) -> Union[
1335        Type[LangfuseAgent],
1336        Type[LangfuseTool],
1337        Type[LangfuseChain],
1338        Type[LangfuseRetriever],
1339        Type[LangfuseEvaluator],
1340        Type[LangfuseEmbedding],
1341        Type[LangfuseGuardrail],
1342        Type[LangfuseGeneration],
1343        Type[LangfuseEvent],
1344        Type[LangfuseSpan],
1345    ]:
1346        """Get the appropriate span class based on as_type."""
1347        normalized_type = as_type.lower()
1348
1349        if normalized_type == "agent":
1350            return LangfuseAgent
1351        elif normalized_type == "tool":
1352            return LangfuseTool
1353        elif normalized_type == "chain":
1354            return LangfuseChain
1355        elif normalized_type == "retriever":
1356            return LangfuseRetriever
1357        elif normalized_type == "evaluator":
1358            return LangfuseEvaluator
1359        elif normalized_type == "embedding":
1360            return LangfuseEmbedding
1361        elif normalized_type == "guardrail":
1362            return LangfuseGuardrail
1363        elif normalized_type == "generation":
1364            return LangfuseGeneration
1365        elif normalized_type == "event":
1366            return LangfuseEvent
1367        elif normalized_type == "span":
1368            return LangfuseSpan
1369        else:
1370            return LangfuseSpan
1371
1372    @_agnosticcontextmanager
1373    def _create_span_with_parent_context(
1374        self,
1375        *,
1376        name: str,
1377        parent: Optional[otel_trace_api.Span] = None,
1378        remote_parent_span: Optional[otel_trace_api.Span] = None,
1379        as_type: ObservationTypeLiteralNoEvent,
1380        end_on_exit: Optional[bool] = None,
1381        input: Optional[Any] = None,
1382        output: Optional[Any] = None,
1383        metadata: Optional[Any] = None,
1384        version: Optional[str] = None,
1385        level: Optional[SpanLevel] = None,
1386        status_message: Optional[str] = None,
1387        completion_start_time: Optional[datetime] = None,
1388        model: Optional[str] = None,
1389        model_parameters: Optional[Dict[str, MapValue]] = None,
1390        usage_details: Optional[Dict[str, int]] = None,
1391        cost_details: Optional[Dict[str, float]] = None,
1392        prompt: Optional[PromptClient] = None,
1393    ) -> Any:
1394        parent_span = parent or cast(otel_trace_api.Span, remote_parent_span)
1395
1396        with otel_trace_api.use_span(parent_span):
1397            with self._start_as_current_otel_span_with_processed_media(
1398                name=name,
1399                as_type=as_type,
1400                end_on_exit=end_on_exit,
1401                input=input,
1402                output=output,
1403                metadata=metadata,
1404                version=version,
1405                level=level,
1406                status_message=status_message,
1407                completion_start_time=completion_start_time,
1408                model=model,
1409                model_parameters=model_parameters,
1410                usage_details=usage_details,
1411                cost_details=cost_details,
1412                prompt=prompt,
1413            ) as langfuse_span:
1414                if remote_parent_span is not None:
1415                    langfuse_span._otel_span.set_attribute(
1416                        LangfuseOtelSpanAttributes.AS_ROOT, True
1417                    )
1418
1419                yield langfuse_span
1420
1421    @_agnosticcontextmanager
1422    def _start_as_current_otel_span_with_processed_media(
1423        self,
1424        *,
1425        name: str,
1426        as_type: Optional[ObservationTypeLiteralNoEvent] = None,
1427        end_on_exit: Optional[bool] = None,
1428        input: Optional[Any] = None,
1429        output: Optional[Any] = None,
1430        metadata: Optional[Any] = None,
1431        version: Optional[str] = None,
1432        level: Optional[SpanLevel] = None,
1433        status_message: Optional[str] = None,
1434        completion_start_time: Optional[datetime] = None,
1435        model: Optional[str] = None,
1436        model_parameters: Optional[Dict[str, MapValue]] = None,
1437        usage_details: Optional[Dict[str, int]] = None,
1438        cost_details: Optional[Dict[str, float]] = None,
1439        prompt: Optional[PromptClient] = None,
1440    ) -> Any:
1441        with self._otel_tracer.start_as_current_span(
1442            name=name,
1443            end_on_exit=end_on_exit if end_on_exit is not None else True,
1444        ) as otel_span:
1445            span_class = self._get_span_class(
1446                as_type or "generation"
1447            )  # default was "generation"
1448            common_args = {
1449                "otel_span": otel_span,
1450                "langfuse_client": self,
1451                "environment": self._environment,
1452                "input": input,
1453                "output": output,
1454                "metadata": metadata,
1455                "version": version,
1456                "level": level,
1457                "status_message": status_message,
1458            }
1459
1460            if span_class in [
1461                LangfuseGeneration,
1462                LangfuseEmbedding,
1463            ]:
1464                common_args.update(
1465                    {
1466                        "completion_start_time": completion_start_time,
1467                        "model": model,
1468                        "model_parameters": model_parameters,
1469                        "usage_details": usage_details,
1470                        "cost_details": cost_details,
1471                        "prompt": prompt,
1472                    }
1473                )
1474            # For span-like types (span, agent, tool, chain, retriever, evaluator, guardrail), no generation properties needed
1475
1476            yield span_class(**common_args)  # type: ignore[arg-type]
1477
1478    def _get_current_otel_span(self) -> Optional[otel_trace_api.Span]:
1479        current_span = otel_trace_api.get_current_span()
1480
1481        if current_span is otel_trace_api.INVALID_SPAN:
1482            langfuse_logger.warning(
1483                "Context error: No active span in current context. Operations that depend on an active span will be skipped. "
1484                "Ensure spans are created with start_as_current_span() or that you're operating within an active span context."
1485            )
1486            return None
1487
1488        return current_span
1489
1490    def update_current_generation(
1491        self,
1492        *,
1493        name: Optional[str] = None,
1494        input: Optional[Any] = None,
1495        output: Optional[Any] = None,
1496        metadata: Optional[Any] = None,
1497        version: Optional[str] = None,
1498        level: Optional[SpanLevel] = None,
1499        status_message: Optional[str] = None,
1500        completion_start_time: Optional[datetime] = None,
1501        model: Optional[str] = None,
1502        model_parameters: Optional[Dict[str, MapValue]] = None,
1503        usage_details: Optional[Dict[str, int]] = None,
1504        cost_details: Optional[Dict[str, float]] = None,
1505        prompt: Optional[PromptClient] = None,
1506    ) -> None:
1507        """Update the current active generation span with new information.
1508
1509        This method updates the current generation span in the active context with
1510        additional information. It's useful for adding output, usage stats, or other
1511        details that become available during or after model generation.
1512
1513        Args:
1514            name: The generation name
1515            input: Updated input data for the model
1516            output: Output from the model (e.g., completions)
1517            metadata: Additional metadata to associate with the generation
1518            version: Version identifier for the model or component
1519            level: Importance level of the generation (info, warning, error)
1520            status_message: Optional status message for the generation
1521            completion_start_time: When the model started generating the response
1522            model: Name/identifier of the AI model used (e.g., "gpt-4")
1523            model_parameters: Parameters used for the model (e.g., temperature, max_tokens)
1524            usage_details: Token usage information (e.g., prompt_tokens, completion_tokens)
1525            cost_details: Cost information for the model call
1526            prompt: Associated prompt template from Langfuse prompt management
1527
1528        Example:
1529            ```python
1530            with langfuse.start_as_current_generation(name="answer-query") as generation:
1531                # Initial setup and API call
1532                response = llm.generate(...)
1533
1534                # Update with results that weren't available at creation time
1535                langfuse.update_current_generation(
1536                    output=response.text,
1537                    usage_details={
1538                        "prompt_tokens": response.usage.prompt_tokens,
1539                        "completion_tokens": response.usage.completion_tokens
1540                    }
1541                )
1542            ```
1543        """
1544        if not self._tracing_enabled:
1545            langfuse_logger.debug(
1546                "Operation skipped: update_current_generation - Tracing is disabled or client is in no-op mode."
1547            )
1548            return
1549
1550        current_otel_span = self._get_current_otel_span()
1551
1552        if current_otel_span is not None:
1553            generation = LangfuseGeneration(
1554                otel_span=current_otel_span, langfuse_client=self
1555            )
1556
1557            if name:
1558                current_otel_span.update_name(name)
1559
1560            generation.update(
1561                input=input,
1562                output=output,
1563                metadata=metadata,
1564                version=version,
1565                level=level,
1566                status_message=status_message,
1567                completion_start_time=completion_start_time,
1568                model=model,
1569                model_parameters=model_parameters,
1570                usage_details=usage_details,
1571                cost_details=cost_details,
1572                prompt=prompt,
1573            )
1574
1575    def update_current_span(
1576        self,
1577        *,
1578        name: Optional[str] = None,
1579        input: Optional[Any] = None,
1580        output: Optional[Any] = None,
1581        metadata: Optional[Any] = None,
1582        version: Optional[str] = None,
1583        level: Optional[SpanLevel] = None,
1584        status_message: Optional[str] = None,
1585    ) -> None:
1586        """Update the current active span with new information.
1587
1588        This method updates the current span in the active context with
1589        additional information. It's useful for adding outputs or metadata
1590        that become available during execution.
1591
1592        Args:
1593            name: The span name
1594            input: Updated input data for the operation
1595            output: Output data from the operation
1596            metadata: Additional metadata to associate with the span
1597            version: Version identifier for the code or component
1598            level: Importance level of the span (info, warning, error)
1599            status_message: Optional status message for the span
1600
1601        Example:
1602            ```python
1603            with langfuse.start_as_current_span(name="process-data") as span:
1604                # Initial processing
1605                result = process_first_part()
1606
1607                # Update with intermediate results
1608                langfuse.update_current_span(metadata={"intermediate_result": result})
1609
1610                # Continue processing
1611                final_result = process_second_part(result)
1612
1613                # Final update
1614                langfuse.update_current_span(output=final_result)
1615            ```
1616        """
1617        if not self._tracing_enabled:
1618            langfuse_logger.debug(
1619                "Operation skipped: update_current_span - Tracing is disabled or client is in no-op mode."
1620            )
1621            return
1622
1623        current_otel_span = self._get_current_otel_span()
1624
1625        if current_otel_span is not None:
1626            span = LangfuseSpan(
1627                otel_span=current_otel_span,
1628                langfuse_client=self,
1629                environment=self._environment,
1630            )
1631
1632            if name:
1633                current_otel_span.update_name(name)
1634
1635            span.update(
1636                input=input,
1637                output=output,
1638                metadata=metadata,
1639                version=version,
1640                level=level,
1641                status_message=status_message,
1642            )
1643
1644    def update_current_trace(
1645        self,
1646        *,
1647        name: Optional[str] = None,
1648        user_id: Optional[str] = None,
1649        session_id: Optional[str] = None,
1650        version: Optional[str] = None,
1651        input: Optional[Any] = None,
1652        output: Optional[Any] = None,
1653        metadata: Optional[Any] = None,
1654        tags: Optional[List[str]] = None,
1655        public: Optional[bool] = None,
1656    ) -> None:
1657        """Update the current trace with additional information.
1658
1659        Args:
1660            name: Updated name for the Langfuse trace
1661            user_id: ID of the user who initiated the Langfuse trace
1662            session_id: Session identifier for grouping related Langfuse traces
1663            version: Version identifier for the application or service
1664            input: Input data for the overall Langfuse trace
1665            output: Output data from the overall Langfuse trace
1666            metadata: Additional metadata to associate with the Langfuse trace
1667            tags: List of tags to categorize the Langfuse trace
1668            public: Whether the Langfuse trace should be publicly accessible
1669
1670        See Also:
1671            :func:`langfuse.propagate_attributes`: Recommended replacement
1672        """
1673        if not self._tracing_enabled:
1674            langfuse_logger.debug(
1675                "Operation skipped: update_current_trace - Tracing is disabled or client is in no-op mode."
1676            )
1677            return
1678
1679        current_otel_span = self._get_current_otel_span()
1680
1681        if current_otel_span is not None and current_otel_span.is_recording():
1682            existing_observation_type = current_otel_span.attributes.get(  # type: ignore[attr-defined]
1683                LangfuseOtelSpanAttributes.OBSERVATION_TYPE, "span"
1684            )
1685            # We need to preserve the class to keep the correct observation type
1686            span_class = self._get_span_class(existing_observation_type)
1687            span = span_class(
1688                otel_span=current_otel_span,
1689                langfuse_client=self,
1690                environment=self._environment,
1691            )
1692
1693            span.update_trace(
1694                name=name,
1695                user_id=user_id,
1696                session_id=session_id,
1697                version=version,
1698                input=input,
1699                output=output,
1700                metadata=metadata,
1701                tags=tags,
1702                public=public,
1703            )
1704
1705    def create_event(
1706        self,
1707        *,
1708        trace_context: Optional[TraceContext] = None,
1709        name: str,
1710        input: Optional[Any] = None,
1711        output: Optional[Any] = None,
1712        metadata: Optional[Any] = None,
1713        version: Optional[str] = None,
1714        level: Optional[SpanLevel] = None,
1715        status_message: Optional[str] = None,
1716    ) -> LangfuseEvent:
1717        """Create a new Langfuse observation of type 'EVENT'.
1718
1719        The created Langfuse Event observation will be the child of the current span in the context.
1720
1721        Args:
1722            trace_context: Optional context for connecting to an existing trace
1723            name: Name of the span (e.g., function or operation name)
1724            input: Input data for the operation (can be any JSON-serializable object)
1725            output: Output data from the operation (can be any JSON-serializable object)
1726            metadata: Additional metadata to associate with the span
1727            version: Version identifier for the code or component
1728            level: Importance level of the span (info, warning, error)
1729            status_message: Optional status message for the span
1730
1731        Returns:
1732            The Langfuse Event object
1733
1734        Example:
1735            ```python
1736            event = langfuse.create_event(name="process-event")
1737            ```
1738        """
1739        timestamp = time_ns()
1740
1741        if trace_context:
1742            trace_id = trace_context.get("trace_id", None)
1743            parent_span_id = trace_context.get("parent_span_id", None)
1744
1745            if trace_id:
1746                remote_parent_span = self._create_remote_parent_span(
1747                    trace_id=trace_id, parent_span_id=parent_span_id
1748                )
1749
1750                with otel_trace_api.use_span(
1751                    cast(otel_trace_api.Span, remote_parent_span)
1752                ):
1753                    otel_span = self._otel_tracer.start_span(
1754                        name=name, start_time=timestamp
1755                    )
1756                    otel_span.set_attribute(LangfuseOtelSpanAttributes.AS_ROOT, True)
1757
1758                    return cast(
1759                        LangfuseEvent,
1760                        LangfuseEvent(
1761                            otel_span=otel_span,
1762                            langfuse_client=self,
1763                            environment=self._environment,
1764                            input=input,
1765                            output=output,
1766                            metadata=metadata,
1767                            version=version,
1768                            level=level,
1769                            status_message=status_message,
1770                        ).end(end_time=timestamp),
1771                    )
1772
1773        otel_span = self._otel_tracer.start_span(name=name, start_time=timestamp)
1774
1775        return cast(
1776            LangfuseEvent,
1777            LangfuseEvent(
1778                otel_span=otel_span,
1779                langfuse_client=self,
1780                environment=self._environment,
1781                input=input,
1782                output=output,
1783                metadata=metadata,
1784                version=version,
1785                level=level,
1786                status_message=status_message,
1787            ).end(end_time=timestamp),
1788        )
1789
1790    def _create_remote_parent_span(
1791        self, *, trace_id: str, parent_span_id: Optional[str]
1792    ) -> Any:
1793        if not self._is_valid_trace_id(trace_id):
1794            langfuse_logger.warning(
1795                f"Passed trace ID '{trace_id}' is not a valid 32 lowercase hex char Langfuse trace id. Ignoring trace ID."
1796            )
1797
1798        if parent_span_id and not self._is_valid_span_id(parent_span_id):
1799            langfuse_logger.warning(
1800                f"Passed span ID '{parent_span_id}' is not a valid 16 lowercase hex char Langfuse span id. Ignoring parent span ID."
1801            )
1802
1803        int_trace_id = int(trace_id, 16)
1804        int_parent_span_id = (
1805            int(parent_span_id, 16)
1806            if parent_span_id
1807            else RandomIdGenerator().generate_span_id()
1808        )
1809
1810        span_context = otel_trace_api.SpanContext(
1811            trace_id=int_trace_id,
1812            span_id=int_parent_span_id,
1813            trace_flags=otel_trace_api.TraceFlags(0x01),  # mark span as sampled
1814            is_remote=False,
1815        )
1816
1817        return otel_trace_api.NonRecordingSpan(span_context)
1818
1819    def _is_valid_trace_id(self, trace_id: str) -> bool:
1820        pattern = r"^[0-9a-f]{32}$"
1821
1822        return bool(re.match(pattern, trace_id))
1823
1824    def _is_valid_span_id(self, span_id: str) -> bool:
1825        pattern = r"^[0-9a-f]{16}$"
1826
1827        return bool(re.match(pattern, span_id))
1828
1829    def _create_observation_id(self, *, seed: Optional[str] = None) -> str:
1830        """Create a unique observation ID for use with Langfuse.
1831
1832        This method generates a unique observation ID (span ID in OpenTelemetry terms)
1833        for use with various Langfuse APIs. It can either generate a random ID or
1834        create a deterministic ID based on a seed string.
1835
1836        Observation IDs must be 16 lowercase hexadecimal characters, representing 8 bytes.
1837        This method ensures the generated ID meets this requirement. If you need to
1838        correlate an external ID with a Langfuse observation ID, use the external ID as
1839        the seed to get a valid, deterministic observation ID.
1840
1841        Args:
1842            seed: Optional string to use as a seed for deterministic ID generation.
1843                 If provided, the same seed will always produce the same ID.
1844                 If not provided, a random ID will be generated.
1845
1846        Returns:
1847            A 16-character lowercase hexadecimal string representing the observation ID.
1848
1849        Example:
1850            ```python
1851            # Generate a random observation ID
1852            obs_id = langfuse.create_observation_id()
1853
1854            # Generate a deterministic ID based on a seed
1855            user_obs_id = langfuse.create_observation_id(seed="user-123-feedback")
1856
1857            # Correlate an external item ID with a Langfuse observation ID
1858            item_id = "item-789012"
1859            correlated_obs_id = langfuse.create_observation_id(seed=item_id)
1860
1861            # Use the ID with Langfuse APIs
1862            langfuse.create_score(
1863                name="relevance",
1864                value=0.95,
1865                trace_id=trace_id,
1866                observation_id=obs_id
1867            )
1868            ```
1869        """
1870        if not seed:
1871            span_id_int = RandomIdGenerator().generate_span_id()
1872
1873            return self._format_otel_span_id(span_id_int)
1874
1875        return sha256(seed.encode("utf-8")).digest()[:8].hex()
1876
1877    @staticmethod
1878    def create_trace_id(*, seed: Optional[str] = None) -> str:
1879        """Create a unique trace ID for use with Langfuse.
1880
1881        This method generates a unique trace ID for use with various Langfuse APIs.
1882        It can either generate a random ID or create a deterministic ID based on
1883        a seed string.
1884
1885        Trace IDs must be 32 lowercase hexadecimal characters, representing 16 bytes.
1886        This method ensures the generated ID meets this requirement. If you need to
1887        correlate an external ID with a Langfuse trace ID, use the external ID as the
1888        seed to get a valid, deterministic Langfuse trace ID.
1889
1890        Args:
1891            seed: Optional string to use as a seed for deterministic ID generation.
1892                 If provided, the same seed will always produce the same ID.
1893                 If not provided, a random ID will be generated.
1894
1895        Returns:
1896            A 32-character lowercase hexadecimal string representing the Langfuse trace ID.
1897
1898        Example:
1899            ```python
1900            # Generate a random trace ID
1901            trace_id = langfuse.create_trace_id()
1902
1903            # Generate a deterministic ID based on a seed
1904            session_trace_id = langfuse.create_trace_id(seed="session-456")
1905
1906            # Correlate an external ID with a Langfuse trace ID
1907            external_id = "external-system-123456"
1908            correlated_trace_id = langfuse.create_trace_id(seed=external_id)
1909
1910            # Use the ID with trace context
1911            with langfuse.start_as_current_span(
1912                name="process-request",
1913                trace_context={"trace_id": trace_id}
1914            ) as span:
1915                # Operation will be part of the specific trace
1916                pass
1917            ```
1918        """
1919        if not seed:
1920            trace_id_int = RandomIdGenerator().generate_trace_id()
1921
1922            return Langfuse._format_otel_trace_id(trace_id_int)
1923
1924        return sha256(seed.encode("utf-8")).digest()[:16].hex()
1925
1926    def _get_otel_trace_id(self, otel_span: otel_trace_api.Span) -> str:
1927        span_context = otel_span.get_span_context()
1928
1929        return self._format_otel_trace_id(span_context.trace_id)
1930
1931    def _get_otel_span_id(self, otel_span: otel_trace_api.Span) -> str:
1932        span_context = otel_span.get_span_context()
1933
1934        return self._format_otel_span_id(span_context.span_id)
1935
1936    @staticmethod
1937    def _format_otel_span_id(span_id_int: int) -> str:
1938        """Format an integer span ID to a 16-character lowercase hex string.
1939
1940        Internal method to convert an OpenTelemetry integer span ID to the standard
1941        W3C Trace Context format (16-character lowercase hex string).
1942
1943        Args:
1944            span_id_int: 64-bit integer representing a span ID
1945
1946        Returns:
1947            A 16-character lowercase hexadecimal string
1948        """
1949        return format(span_id_int, "016x")
1950
1951    @staticmethod
1952    def _format_otel_trace_id(trace_id_int: int) -> str:
1953        """Format an integer trace ID to a 32-character lowercase hex string.
1954
1955        Internal method to convert an OpenTelemetry integer trace ID to the standard
1956        W3C Trace Context format (32-character lowercase hex string).
1957
1958        Args:
1959            trace_id_int: 128-bit integer representing a trace ID
1960
1961        Returns:
1962            A 32-character lowercase hexadecimal string
1963        """
1964        return format(trace_id_int, "032x")
1965
1966    @overload
1967    def create_score(
1968        self,
1969        *,
1970        name: str,
1971        value: float,
1972        session_id: Optional[str] = None,
1973        dataset_run_id: Optional[str] = None,
1974        trace_id: Optional[str] = None,
1975        observation_id: Optional[str] = None,
1976        score_id: Optional[str] = None,
1977        data_type: Optional[Literal["NUMERIC", "BOOLEAN"]] = None,
1978        comment: Optional[str] = None,
1979        config_id: Optional[str] = None,
1980        metadata: Optional[Any] = None,
1981        timestamp: Optional[datetime] = None,
1982    ) -> None: ...
1983
1984    @overload
1985    def create_score(
1986        self,
1987        *,
1988        name: str,
1989        value: str,
1990        session_id: Optional[str] = None,
1991        dataset_run_id: Optional[str] = None,
1992        trace_id: Optional[str] = None,
1993        score_id: Optional[str] = None,
1994        observation_id: Optional[str] = None,
1995        data_type: Optional[Literal["CATEGORICAL"]] = "CATEGORICAL",
1996        comment: Optional[str] = None,
1997        config_id: Optional[str] = None,
1998        metadata: Optional[Any] = None,
1999        timestamp: Optional[datetime] = None,
2000    ) -> None: ...
2001
2002    def create_score(
2003        self,
2004        *,
2005        name: str,
2006        value: Union[float, str],
2007        session_id: Optional[str] = None,
2008        dataset_run_id: Optional[str] = None,
2009        trace_id: Optional[str] = None,
2010        observation_id: Optional[str] = None,
2011        score_id: Optional[str] = None,
2012        data_type: Optional[ScoreDataType] = None,
2013        comment: Optional[str] = None,
2014        config_id: Optional[str] = None,
2015        metadata: Optional[Any] = None,
2016        timestamp: Optional[datetime] = None,
2017    ) -> None:
2018        """Create a score for a specific trace or observation.
2019
2020        This method creates a score for evaluating a Langfuse trace or observation. Scores can be
2021        used to track quality metrics, user feedback, or automated evaluations.
2022
2023        Args:
2024            name: Name of the score (e.g., "relevance", "accuracy")
2025            value: Score value (can be numeric for NUMERIC/BOOLEAN types or string for CATEGORICAL)
2026            session_id: ID of the Langfuse session to associate the score with
2027            dataset_run_id: ID of the Langfuse dataset run to associate the score with
2028            trace_id: ID of the Langfuse trace to associate the score with
2029            observation_id: Optional ID of the specific observation to score. Trace ID must be provided too.
2030            score_id: Optional custom ID for the score (auto-generated if not provided)
2031            data_type: Type of score (NUMERIC, BOOLEAN, or CATEGORICAL)
2032            comment: Optional comment or explanation for the score
2033            config_id: Optional ID of a score config defined in Langfuse
2034            metadata: Optional metadata to be attached to the score
2035            timestamp: Optional timestamp for the score (defaults to current UTC time)
2036
2037        Example:
2038            ```python
2039            # Create a numeric score for accuracy
2040            langfuse.create_score(
2041                name="accuracy",
2042                value=0.92,
2043                trace_id="abcdef1234567890abcdef1234567890",
2044                data_type="NUMERIC",
2045                comment="High accuracy with minor irrelevant details"
2046            )
2047
2048            # Create a categorical score for sentiment
2049            langfuse.create_score(
2050                name="sentiment",
2051                value="positive",
2052                trace_id="abcdef1234567890abcdef1234567890",
2053                observation_id="abcdef1234567890",
2054                data_type="CATEGORICAL"
2055            )
2056            ```
2057        """
2058        if not self._tracing_enabled:
2059            return
2060
2061        score_id = score_id or self._create_observation_id()
2062
2063        try:
2064            new_body = ScoreBody(
2065                id=score_id,
2066                sessionId=session_id,
2067                datasetRunId=dataset_run_id,
2068                traceId=trace_id,
2069                observationId=observation_id,
2070                name=name,
2071                value=value,
2072                dataType=data_type,  # type: ignore
2073                comment=comment,
2074                configId=config_id,
2075                environment=self._environment,
2076                metadata=metadata,
2077            )
2078
2079            event = {
2080                "id": self.create_trace_id(),
2081                "type": "score-create",
2082                "timestamp": timestamp or _get_timestamp(),
2083                "body": new_body,
2084            }
2085
2086            if self._resources is not None:
2087                # Force the score to be in sample if it was for a legacy trace ID, i.e. non-32 hexchar
2088                force_sample = (
2089                    not self._is_valid_trace_id(trace_id) if trace_id else True
2090                )
2091
2092                self._resources.add_score_task(
2093                    event,
2094                    force_sample=force_sample,
2095                )
2096
2097        except Exception as e:
2098            langfuse_logger.exception(
2099                f"Error creating score: Failed to process score event for trace_id={trace_id}, name={name}. Error: {e}"
2100            )
2101
2102    @overload
2103    def score_current_span(
2104        self,
2105        *,
2106        name: str,
2107        value: float,
2108        score_id: Optional[str] = None,
2109        data_type: Optional[Literal["NUMERIC", "BOOLEAN"]] = None,
2110        comment: Optional[str] = None,
2111        config_id: Optional[str] = None,
2112        metadata: Optional[Any] = None,
2113    ) -> None: ...
2114
2115    @overload
2116    def score_current_span(
2117        self,
2118        *,
2119        name: str,
2120        value: str,
2121        score_id: Optional[str] = None,
2122        data_type: Optional[Literal["CATEGORICAL"]] = "CATEGORICAL",
2123        comment: Optional[str] = None,
2124        config_id: Optional[str] = None,
2125        metadata: Optional[Any] = None,
2126    ) -> None: ...
2127
2128    def score_current_span(
2129        self,
2130        *,
2131        name: str,
2132        value: Union[float, str],
2133        score_id: Optional[str] = None,
2134        data_type: Optional[ScoreDataType] = None,
2135        comment: Optional[str] = None,
2136        config_id: Optional[str] = None,
2137        metadata: Optional[Any] = None,
2138    ) -> None:
2139        """Create a score for the current active span.
2140
2141        This method scores the currently active span in the context. It's a convenient
2142        way to score the current operation without needing to know its trace and span IDs.
2143
2144        Args:
2145            name: Name of the score (e.g., "relevance", "accuracy")
2146            value: Score value (can be numeric for NUMERIC/BOOLEAN types or string for CATEGORICAL)
2147            score_id: Optional custom ID for the score (auto-generated if not provided)
2148            data_type: Type of score (NUMERIC, BOOLEAN, or CATEGORICAL)
2149            comment: Optional comment or explanation for the score
2150            config_id: Optional ID of a score config defined in Langfuse
2151            metadata: Optional metadata to be attached to the score
2152
2153        Example:
2154            ```python
2155            with langfuse.start_as_current_generation(name="answer-query") as generation:
2156                # Generate answer
2157                response = generate_answer(...)
2158                generation.update(output=response)
2159
2160                # Score the generation
2161                langfuse.score_current_span(
2162                    name="relevance",
2163                    value=0.85,
2164                    data_type="NUMERIC",
2165                    comment="Mostly relevant but contains some tangential information",
2166                    metadata={"model": "gpt-4", "prompt_version": "v2"}
2167                )
2168            ```
2169        """
2170        current_span = self._get_current_otel_span()
2171
2172        if current_span is not None:
2173            trace_id = self._get_otel_trace_id(current_span)
2174            observation_id = self._get_otel_span_id(current_span)
2175
2176            langfuse_logger.info(
2177                f"Score: Creating score name='{name}' value={value} for current span ({observation_id}) in trace {trace_id}"
2178            )
2179
2180            self.create_score(
2181                trace_id=trace_id,
2182                observation_id=observation_id,
2183                name=name,
2184                value=cast(str, value),
2185                score_id=score_id,
2186                data_type=cast(Literal["CATEGORICAL"], data_type),
2187                comment=comment,
2188                config_id=config_id,
2189                metadata=metadata,
2190            )
2191
2192    @overload
2193    def score_current_trace(
2194        self,
2195        *,
2196        name: str,
2197        value: float,
2198        score_id: Optional[str] = None,
2199        data_type: Optional[Literal["NUMERIC", "BOOLEAN"]] = None,
2200        comment: Optional[str] = None,
2201        config_id: Optional[str] = None,
2202        metadata: Optional[Any] = None,
2203    ) -> None: ...
2204
2205    @overload
2206    def score_current_trace(
2207        self,
2208        *,
2209        name: str,
2210        value: str,
2211        score_id: Optional[str] = None,
2212        data_type: Optional[Literal["CATEGORICAL"]] = "CATEGORICAL",
2213        comment: Optional[str] = None,
2214        config_id: Optional[str] = None,
2215        metadata: Optional[Any] = None,
2216    ) -> None: ...
2217
2218    def score_current_trace(
2219        self,
2220        *,
2221        name: str,
2222        value: Union[float, str],
2223        score_id: Optional[str] = None,
2224        data_type: Optional[ScoreDataType] = None,
2225        comment: Optional[str] = None,
2226        config_id: Optional[str] = None,
2227        metadata: Optional[Any] = None,
2228    ) -> None:
2229        """Create a score for the current trace.
2230
2231        This method scores the trace of the currently active span. Unlike score_current_span,
2232        this method associates the score with the entire trace rather than a specific span.
2233        It's useful for scoring overall performance or quality of the entire operation.
2234
2235        Args:
2236            name: Name of the score (e.g., "user_satisfaction", "overall_quality")
2237            value: Score value (can be numeric for NUMERIC/BOOLEAN types or string for CATEGORICAL)
2238            score_id: Optional custom ID for the score (auto-generated if not provided)
2239            data_type: Type of score (NUMERIC, BOOLEAN, or CATEGORICAL)
2240            comment: Optional comment or explanation for the score
2241            config_id: Optional ID of a score config defined in Langfuse
2242            metadata: Optional metadata to be attached to the score
2243
2244        Example:
2245            ```python
2246            with langfuse.start_as_current_span(name="process-user-request") as span:
2247                # Process request
2248                result = process_complete_request()
2249                span.update(output=result)
2250
2251                # Score the overall trace
2252                langfuse.score_current_trace(
2253                    name="overall_quality",
2254                    value=0.95,
2255                    data_type="NUMERIC",
2256                    comment="High quality end-to-end response",
2257                    metadata={"evaluator": "gpt-4", "criteria": "comprehensive"}
2258                )
2259            ```
2260        """
2261        current_span = self._get_current_otel_span()
2262
2263        if current_span is not None:
2264            trace_id = self._get_otel_trace_id(current_span)
2265
2266            langfuse_logger.info(
2267                f"Score: Creating score name='{name}' value={value} for entire trace {trace_id}"
2268            )
2269
2270            self.create_score(
2271                trace_id=trace_id,
2272                name=name,
2273                value=cast(str, value),
2274                score_id=score_id,
2275                data_type=cast(Literal["CATEGORICAL"], data_type),
2276                comment=comment,
2277                config_id=config_id,
2278                metadata=metadata,
2279            )
2280
2281    def flush(self) -> None:
2282        """Force flush all pending spans and events to the Langfuse API.
2283
2284        This method manually flushes any pending spans, scores, and other events to the
2285        Langfuse API. It's useful in scenarios where you want to ensure all data is sent
2286        before proceeding, without waiting for the automatic flush interval.
2287
2288        Example:
2289            ```python
2290            # Record some spans and scores
2291            with langfuse.start_as_current_span(name="operation") as span:
2292                # Do work...
2293                pass
2294
2295            # Ensure all data is sent to Langfuse before proceeding
2296            langfuse.flush()
2297
2298            # Continue with other work
2299            ```
2300        """
2301        if self._resources is not None:
2302            self._resources.flush()
2303
2304    def shutdown(self) -> None:
2305        """Shut down the Langfuse client and flush all pending data.
2306
2307        This method cleanly shuts down the Langfuse client, ensuring all pending data
2308        is flushed to the API and all background threads are properly terminated.
2309
2310        It's important to call this method when your application is shutting down to
2311        prevent data loss and resource leaks. For most applications, using the client
2312        as a context manager or relying on the automatic shutdown via atexit is sufficient.
2313
2314        Example:
2315            ```python
2316            # Initialize Langfuse
2317            langfuse = Langfuse(public_key="...", secret_key="...")
2318
2319            # Use Langfuse throughout your application
2320            # ...
2321
2322            # When application is shutting down
2323            langfuse.shutdown()
2324            ```
2325        """
2326        if self._resources is not None:
2327            self._resources.shutdown()
2328
2329    def get_current_trace_id(self) -> Optional[str]:
2330        """Get the trace ID of the current active span.
2331
2332        This method retrieves the trace ID from the currently active span in the context.
2333        It can be used to get the trace ID for referencing in logs, external systems,
2334        or for creating related operations.
2335
2336        Returns:
2337            The current trace ID as a 32-character lowercase hexadecimal string,
2338            or None if there is no active span.
2339
2340        Example:
2341            ```python
2342            with langfuse.start_as_current_span(name="process-request") as span:
2343                # Get the current trace ID for reference
2344                trace_id = langfuse.get_current_trace_id()
2345
2346                # Use it for external correlation
2347                log.info(f"Processing request with trace_id: {trace_id}")
2348
2349                # Or pass to another system
2350                external_system.process(data, trace_id=trace_id)
2351            ```
2352        """
2353        if not self._tracing_enabled:
2354            langfuse_logger.debug(
2355                "Operation skipped: get_current_trace_id - Tracing is disabled or client is in no-op mode."
2356            )
2357            return None
2358
2359        current_otel_span = self._get_current_otel_span()
2360
2361        return self._get_otel_trace_id(current_otel_span) if current_otel_span else None
2362
2363    def get_current_observation_id(self) -> Optional[str]:
2364        """Get the observation ID (span ID) of the current active span.
2365
2366        This method retrieves the observation ID from the currently active span in the context.
2367        It can be used to get the observation ID for referencing in logs, external systems,
2368        or for creating scores or other related operations.
2369
2370        Returns:
2371            The current observation ID as a 16-character lowercase hexadecimal string,
2372            or None if there is no active span.
2373
2374        Example:
2375            ```python
2376            with langfuse.start_as_current_span(name="process-user-query") as span:
2377                # Get the current observation ID
2378                observation_id = langfuse.get_current_observation_id()
2379
2380                # Store it for later reference
2381                cache.set(f"query_{query_id}_observation", observation_id)
2382
2383                # Process the query...
2384            ```
2385        """
2386        if not self._tracing_enabled:
2387            langfuse_logger.debug(
2388                "Operation skipped: get_current_observation_id - Tracing is disabled or client is in no-op mode."
2389            )
2390            return None
2391
2392        current_otel_span = self._get_current_otel_span()
2393
2394        return self._get_otel_span_id(current_otel_span) if current_otel_span else None
2395
2396    def _get_project_id(self) -> Optional[str]:
2397        """Fetch and return the current project id. Persisted across requests. Returns None if no project id is found for api keys."""
2398        if not self._project_id:
2399            proj = self.api.projects.get()
2400            if not proj.data or not proj.data[0].id:
2401                return None
2402
2403            self._project_id = proj.data[0].id
2404
2405        return self._project_id
2406
2407    def get_trace_url(self, *, trace_id: Optional[str] = None) -> Optional[str]:
2408        """Get the URL to view a trace in the Langfuse UI.
2409
2410        This method generates a URL that links directly to a trace in the Langfuse UI.
2411        It's useful for providing links in logs, notifications, or debugging tools.
2412
2413        Args:
2414            trace_id: Optional trace ID to generate a URL for. If not provided,
2415                     the trace ID of the current active span will be used.
2416
2417        Returns:
2418            A URL string pointing to the trace in the Langfuse UI,
2419            or None if the project ID couldn't be retrieved or no trace ID is available.
2420
2421        Example:
2422            ```python
2423            # Get URL for the current trace
2424            with langfuse.start_as_current_span(name="process-request") as span:
2425                trace_url = langfuse.get_trace_url()
2426                log.info(f"Processing trace: {trace_url}")
2427
2428            # Get URL for a specific trace
2429            specific_trace_url = langfuse.get_trace_url(trace_id="1234567890abcdef1234567890abcdef")
2430            send_notification(f"Review needed for trace: {specific_trace_url}")
2431            ```
2432        """
2433        final_trace_id = trace_id or self.get_current_trace_id()
2434        if not final_trace_id:
2435            return None
2436
2437        project_id = self._get_project_id()
2438
2439        return (
2440            f"{self._base_url}/project/{project_id}/traces/{final_trace_id}"
2441            if project_id and final_trace_id
2442            else None
2443        )
2444
2445    def get_dataset(
2446        self, name: str, *, fetch_items_page_size: Optional[int] = 50
2447    ) -> "DatasetClient":
2448        """Fetch a dataset by its name.
2449
2450        Args:
2451            name (str): The name of the dataset to fetch.
2452            fetch_items_page_size (Optional[int]): All items of the dataset will be fetched in chunks of this size. Defaults to 50.
2453
2454        Returns:
2455            DatasetClient: The dataset with the given name.
2456        """
2457        try:
2458            langfuse_logger.debug(f"Getting datasets {name}")
2459            dataset = self.api.datasets.get(dataset_name=self._url_encode(name))
2460
2461            dataset_items = []
2462            page = 1
2463
2464            while True:
2465                new_items = self.api.dataset_items.list(
2466                    dataset_name=self._url_encode(name, is_url_param=True),
2467                    page=page,
2468                    limit=fetch_items_page_size,
2469                )
2470                dataset_items.extend(new_items.data)
2471
2472                if new_items.meta.total_pages <= page:
2473                    break
2474
2475                page += 1
2476
2477            items = [DatasetItemClient(i, langfuse=self) for i in dataset_items]
2478
2479            return DatasetClient(dataset, items=items)
2480
2481        except Error as e:
2482            handle_fern_exception(e)
2483            raise e
2484
2485    def get_dataset_run(
2486        self, *, dataset_name: str, run_name: str
2487    ) -> DatasetRunWithItems:
2488        """Fetch a dataset run by dataset name and run name.
2489
2490        Args:
2491            dataset_name (str): The name of the dataset.
2492            run_name (str): The name of the run.
2493
2494        Returns:
2495            DatasetRunWithItems: The dataset run with its items.
2496        """
2497        try:
2498            return self.api.datasets.get_run(
2499                dataset_name=self._url_encode(dataset_name),
2500                run_name=self._url_encode(run_name),
2501                request_options=None,
2502            )
2503        except Error as e:
2504            handle_fern_exception(e)
2505            raise e
2506
2507    def get_dataset_runs(
2508        self,
2509        *,
2510        dataset_name: str,
2511        page: Optional[int] = None,
2512        limit: Optional[int] = None,
2513    ) -> PaginatedDatasetRuns:
2514        """Fetch all runs for a dataset.
2515
2516        Args:
2517            dataset_name (str): The name of the dataset.
2518            page (Optional[int]): Page number, starts at 1.
2519            limit (Optional[int]): Limit of items per page.
2520
2521        Returns:
2522            PaginatedDatasetRuns: Paginated list of dataset runs.
2523        """
2524        try:
2525            return self.api.datasets.get_runs(
2526                dataset_name=self._url_encode(dataset_name),
2527                page=page,
2528                limit=limit,
2529                request_options=None,
2530            )
2531        except Error as e:
2532            handle_fern_exception(e)
2533            raise e
2534
2535    def delete_dataset_run(
2536        self, *, dataset_name: str, run_name: str
2537    ) -> DeleteDatasetRunResponse:
2538        """Delete a dataset run and all its run items. This action is irreversible.
2539
2540        Args:
2541            dataset_name (str): The name of the dataset.
2542            run_name (str): The name of the run.
2543
2544        Returns:
2545            DeleteDatasetRunResponse: Confirmation of deletion.
2546        """
2547        try:
2548            return self.api.datasets.delete_run(
2549                dataset_name=self._url_encode(dataset_name),
2550                run_name=self._url_encode(run_name),
2551                request_options=None,
2552            )
2553        except Error as e:
2554            handle_fern_exception(e)
2555            raise e
2556
2557    def run_experiment(
2558        self,
2559        *,
2560        name: str,
2561        run_name: Optional[str] = None,
2562        description: Optional[str] = None,
2563        data: ExperimentData,
2564        task: TaskFunction,
2565        evaluators: List[EvaluatorFunction] = [],
2566        composite_evaluator: Optional[CompositeEvaluatorFunction] = None,
2567        run_evaluators: List[RunEvaluatorFunction] = [],
2568        max_concurrency: int = 50,
2569        metadata: Optional[Dict[str, str]] = None,
2570    ) -> ExperimentResult:
2571        """Run an experiment on a dataset with automatic tracing and evaluation.
2572
2573        This method executes a task function on each item in the provided dataset,
2574        automatically traces all executions with Langfuse for observability, runs
2575        item-level and run-level evaluators on the outputs, and returns comprehensive
2576        results with evaluation metrics.
2577
2578        The experiment system provides:
2579        - Automatic tracing of all task executions
2580        - Concurrent processing with configurable limits
2581        - Comprehensive error handling that isolates failures
2582        - Integration with Langfuse datasets for experiment tracking
2583        - Flexible evaluation framework supporting both sync and async evaluators
2584
2585        Args:
2586            name: Human-readable name for the experiment. Used for identification
2587                in the Langfuse UI.
2588            run_name: Optional exact name for the experiment run. If provided, this will be
2589                used as the exact dataset run name if the `data` contains Langfuse dataset items.
2590                If not provided, this will default to the experiment name appended with an ISO timestamp.
2591            description: Optional description explaining the experiment's purpose,
2592                methodology, or expected outcomes.
2593            data: Array of data items to process. Can be either:
2594                - List of dict-like items with 'input', 'expected_output', 'metadata' keys
2595                - List of Langfuse DatasetItem objects from dataset.items
2596            task: Function that processes each data item and returns output.
2597                Must accept 'item' as keyword argument and can return sync or async results.
2598                The task function signature should be: task(*, item, **kwargs) -> Any
2599            evaluators: List of functions to evaluate each item's output individually.
2600                Each evaluator receives input, output, expected_output, and metadata.
2601                Can return single Evaluation dict or list of Evaluation dicts.
2602            composite_evaluator: Optional function that creates composite scores from item-level evaluations.
2603                Receives the same inputs as item-level evaluators (input, output, expected_output, metadata)
2604                plus the list of evaluations from item-level evaluators. Useful for weighted averages,
2605                pass/fail decisions based on multiple criteria, or custom scoring logic combining multiple metrics.
2606            run_evaluators: List of functions to evaluate the entire experiment run.
2607                Each run evaluator receives all item_results and can compute aggregate metrics.
2608                Useful for calculating averages, distributions, or cross-item comparisons.
2609            max_concurrency: Maximum number of concurrent task executions (default: 50).
2610                Controls the number of items processed simultaneously. Adjust based on
2611                API rate limits and system resources.
2612            metadata: Optional metadata dictionary to attach to all experiment traces.
2613                This metadata will be included in every trace created during the experiment.
2614                If `data` are Langfuse dataset items, the metadata will be attached to the dataset run, too.
2615
2616        Returns:
2617            ExperimentResult containing:
2618            - run_name: The experiment run name. This is equal to the dataset run name if experiment was on Langfuse dataset.
2619            - item_results: List of results for each processed item with outputs and evaluations
2620            - run_evaluations: List of aggregate evaluation results for the entire run
2621            - dataset_run_id: ID of the dataset run (if using Langfuse datasets)
2622            - dataset_run_url: Direct URL to view results in Langfuse UI (if applicable)
2623
2624        Raises:
2625            ValueError: If required parameters are missing or invalid
2626            Exception: If experiment setup fails (individual item failures are handled gracefully)
2627
2628        Examples:
2629            Basic experiment with local data:
2630            ```python
2631            def summarize_text(*, item, **kwargs):
2632                return f"Summary: {item['input'][:50]}..."
2633
2634            def length_evaluator(*, input, output, expected_output=None, **kwargs):
2635                return {
2636                    "name": "output_length",
2637                    "value": len(output),
2638                    "comment": f"Output contains {len(output)} characters"
2639                }
2640
2641            result = langfuse.run_experiment(
2642                name="Text Summarization Test",
2643                description="Evaluate summarization quality and length",
2644                data=[
2645                    {"input": "Long article text...", "expected_output": "Expected summary"},
2646                    {"input": "Another article...", "expected_output": "Another summary"}
2647                ],
2648                task=summarize_text,
2649                evaluators=[length_evaluator]
2650            )
2651
2652            print(f"Processed {len(result.item_results)} items")
2653            for item_result in result.item_results:
2654                print(f"Input: {item_result.item['input']}")
2655                print(f"Output: {item_result.output}")
2656                print(f"Evaluations: {item_result.evaluations}")
2657            ```
2658
2659            Advanced experiment with async task and multiple evaluators:
2660            ```python
2661            async def llm_task(*, item, **kwargs):
2662                # Simulate async LLM call
2663                response = await openai_client.chat.completions.create(
2664                    model="gpt-4",
2665                    messages=[{"role": "user", "content": item["input"]}]
2666                )
2667                return response.choices[0].message.content
2668
2669            def accuracy_evaluator(*, input, output, expected_output=None, **kwargs):
2670                if expected_output and expected_output.lower() in output.lower():
2671                    return {"name": "accuracy", "value": 1.0, "comment": "Correct answer"}
2672                return {"name": "accuracy", "value": 0.0, "comment": "Incorrect answer"}
2673
2674            def toxicity_evaluator(*, input, output, expected_output=None, **kwargs):
2675                # Simulate toxicity check
2676                toxicity_score = check_toxicity(output)  # Your toxicity checker
2677                return {
2678                    "name": "toxicity",
2679                    "value": toxicity_score,
2680                    "comment": f"Toxicity level: {'high' if toxicity_score > 0.7 else 'low'}"
2681                }
2682
2683            def average_accuracy(*, item_results, **kwargs):
2684                accuracies = [
2685                    eval.value for result in item_results
2686                    for eval in result.evaluations
2687                    if eval.name == "accuracy"
2688                ]
2689                return {
2690                    "name": "average_accuracy",
2691                    "value": sum(accuracies) / len(accuracies) if accuracies else 0,
2692                    "comment": f"Average accuracy across {len(accuracies)} items"
2693                }
2694
2695            result = langfuse.run_experiment(
2696                name="LLM Safety and Accuracy Test",
2697                description="Evaluate model accuracy and safety across diverse prompts",
2698                data=test_dataset,  # Your dataset items
2699                task=llm_task,
2700                evaluators=[accuracy_evaluator, toxicity_evaluator],
2701                run_evaluators=[average_accuracy],
2702                max_concurrency=5,  # Limit concurrent API calls
2703                metadata={"model": "gpt-4", "temperature": 0.7}
2704            )
2705            ```
2706
2707            Using with Langfuse datasets:
2708            ```python
2709            # Get dataset from Langfuse
2710            dataset = langfuse.get_dataset("my-eval-dataset")
2711
2712            result = dataset.run_experiment(
2713                name="Production Model Evaluation",
2714                description="Monthly evaluation of production model performance",
2715                task=my_production_task,
2716                evaluators=[accuracy_evaluator, latency_evaluator]
2717            )
2718
2719            # Results automatically linked to dataset in Langfuse UI
2720            print(f"View results: {result['dataset_run_url']}")
2721            ```
2722
2723        Note:
2724            - Task and evaluator functions can be either synchronous or asynchronous
2725            - Individual item failures are logged but don't stop the experiment
2726            - All executions are automatically traced and visible in Langfuse UI
2727            - When using Langfuse datasets, results are automatically linked for easy comparison
2728            - This method works in both sync and async contexts (Jupyter notebooks, web apps, etc.)
2729            - Async execution is handled automatically with smart event loop detection
2730        """
2731        return cast(
2732            ExperimentResult,
2733            run_async_safely(
2734                self._run_experiment_async(
2735                    name=name,
2736                    run_name=self._create_experiment_run_name(
2737                        name=name, run_name=run_name
2738                    ),
2739                    description=description,
2740                    data=data,
2741                    task=task,
2742                    evaluators=evaluators or [],
2743                    composite_evaluator=composite_evaluator,
2744                    run_evaluators=run_evaluators or [],
2745                    max_concurrency=max_concurrency,
2746                    metadata=metadata,
2747                ),
2748            ),
2749        )
2750
2751    async def _run_experiment_async(
2752        self,
2753        *,
2754        name: str,
2755        run_name: str,
2756        description: Optional[str],
2757        data: ExperimentData,
2758        task: TaskFunction,
2759        evaluators: List[EvaluatorFunction],
2760        composite_evaluator: Optional[CompositeEvaluatorFunction],
2761        run_evaluators: List[RunEvaluatorFunction],
2762        max_concurrency: int,
2763        metadata: Optional[Dict[str, Any]] = None,
2764    ) -> ExperimentResult:
2765        langfuse_logger.debug(
2766            f"Starting experiment '{name}' run '{run_name}' with {len(data)} items"
2767        )
2768
2769        # Set up concurrency control
2770        semaphore = asyncio.Semaphore(max_concurrency)
2771
2772        # Process all items
2773        async def process_item(item: ExperimentItem) -> ExperimentItemResult:
2774            async with semaphore:
2775                return await self._process_experiment_item(
2776                    item,
2777                    task,
2778                    evaluators,
2779                    composite_evaluator,
2780                    name,
2781                    run_name,
2782                    description,
2783                    metadata,
2784                )
2785
2786        # Run all items concurrently
2787        tasks = [process_item(item) for item in data]
2788        item_results = await asyncio.gather(*tasks, return_exceptions=True)
2789
2790        # Filter out any exceptions and log errors
2791        valid_results: List[ExperimentItemResult] = []
2792        for i, result in enumerate(item_results):
2793            if isinstance(result, Exception):
2794                langfuse_logger.error(f"Item {i} failed: {result}")
2795            elif isinstance(result, ExperimentItemResult):
2796                valid_results.append(result)  # type: ignore
2797
2798        # Run experiment-level evaluators
2799        run_evaluations: List[Evaluation] = []
2800        for run_evaluator in run_evaluators:
2801            try:
2802                evaluations = await _run_evaluator(
2803                    run_evaluator, item_results=valid_results
2804                )
2805                run_evaluations.extend(evaluations)
2806            except Exception as e:
2807                langfuse_logger.error(f"Run evaluator failed: {e}")
2808
2809        # Generate dataset run URL if applicable
2810        dataset_run_id = valid_results[0].dataset_run_id if valid_results else None
2811        dataset_run_url = None
2812        if dataset_run_id and data:
2813            try:
2814                # Check if the first item has dataset_id (for DatasetItem objects)
2815                first_item = data[0]
2816                dataset_id = None
2817
2818                if hasattr(first_item, "dataset_id"):
2819                    dataset_id = getattr(first_item, "dataset_id", None)
2820
2821                if dataset_id:
2822                    project_id = self._get_project_id()
2823
2824                    if project_id:
2825                        dataset_run_url = f"{self._base_url}/project/{project_id}/datasets/{dataset_id}/runs/{dataset_run_id}"
2826
2827            except Exception:
2828                pass  # URL generation is optional
2829
2830        # Store run-level evaluations as scores
2831        for evaluation in run_evaluations:
2832            try:
2833                if dataset_run_id:
2834                    self.create_score(
2835                        dataset_run_id=dataset_run_id,
2836                        name=evaluation.name or "<unknown>",
2837                        value=evaluation.value,  # type: ignore
2838                        comment=evaluation.comment,
2839                        metadata=evaluation.metadata,
2840                        data_type=evaluation.data_type,  # type: ignore
2841                        config_id=evaluation.config_id,
2842                    )
2843
2844            except Exception as e:
2845                langfuse_logger.error(f"Failed to store run evaluation: {e}")
2846
2847        # Flush scores and traces
2848        self.flush()
2849
2850        return ExperimentResult(
2851            name=name,
2852            run_name=run_name,
2853            description=description,
2854            item_results=valid_results,
2855            run_evaluations=run_evaluations,
2856            dataset_run_id=dataset_run_id,
2857            dataset_run_url=dataset_run_url,
2858        )
2859
2860    async def _process_experiment_item(
2861        self,
2862        item: ExperimentItem,
2863        task: Callable,
2864        evaluators: List[Callable],
2865        composite_evaluator: Optional[CompositeEvaluatorFunction],
2866        experiment_name: str,
2867        experiment_run_name: str,
2868        experiment_description: Optional[str],
2869        experiment_metadata: Optional[Dict[str, Any]] = None,
2870    ) -> ExperimentItemResult:
2871        span_name = "experiment-item-run"
2872
2873        with self.start_as_current_span(name=span_name) as span:
2874            try:
2875                input_data = (
2876                    item.get("input")
2877                    if isinstance(item, dict)
2878                    else getattr(item, "input", None)
2879                )
2880
2881                if input_data is None:
2882                    raise ValueError("Experiment Item is missing input. Skipping item.")
2883
2884                expected_output = (
2885                    item.get("expected_output")
2886                    if isinstance(item, dict)
2887                    else getattr(item, "expected_output", None)
2888                )
2889
2890                item_metadata = (
2891                    item.get("metadata")
2892                    if isinstance(item, dict)
2893                    else getattr(item, "metadata", None)
2894                )
2895
2896                final_observation_metadata = {
2897                    "experiment_name": experiment_name,
2898                    "experiment_run_name": experiment_run_name,
2899                    **(experiment_metadata or {}),
2900                }
2901
2902                trace_id = span.trace_id
2903                dataset_id = None
2904                dataset_item_id = None
2905                dataset_run_id = None
2906
2907                # Link to dataset run if this is a dataset item
2908                if hasattr(item, "id") and hasattr(item, "dataset_id"):
2909                    try:
2910                        # Use sync API to avoid event loop issues when run_async_safely
2911                        # creates multiple event loops across different threads
2912                        dataset_run_item = await asyncio.to_thread(
2913                            self.api.dataset_run_items.create,
2914                            request=CreateDatasetRunItemRequest(
2915                                runName=experiment_run_name,
2916                                runDescription=experiment_description,
2917                                metadata=experiment_metadata,
2918                                datasetItemId=item.id,  # type: ignore
2919                                traceId=trace_id,
2920                                observationId=span.id,
2921                            ),
2922                        )
2923
2924                        dataset_run_id = dataset_run_item.dataset_run_id
2925
2926                    except Exception as e:
2927                        langfuse_logger.error(f"Failed to create dataset run item: {e}")
2928
2929                if (
2930                    not isinstance(item, dict)
2931                    and hasattr(item, "dataset_id")
2932                    and hasattr(item, "id")
2933                ):
2934                    dataset_id = item.dataset_id
2935                    dataset_item_id = item.id
2936
2937                    final_observation_metadata.update(
2938                        {"dataset_id": dataset_id, "dataset_item_id": dataset_item_id}
2939                    )
2940
2941                if isinstance(item_metadata, dict):
2942                    final_observation_metadata.update(item_metadata)
2943
2944                experiment_id = dataset_run_id or self._create_observation_id()
2945                experiment_item_id = (
2946                    dataset_item_id or get_sha256_hash_hex(_serialize(input_data))[:16]
2947                )
2948                span._otel_span.set_attributes(
2949                    {
2950                        k: v
2951                        for k, v in {
2952                            LangfuseOtelSpanAttributes.ENVIRONMENT: LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT,
2953                            LangfuseOtelSpanAttributes.EXPERIMENT_DESCRIPTION: experiment_description,
2954                            LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_EXPECTED_OUTPUT: _serialize(
2955                                expected_output
2956                            ),
2957                        }.items()
2958                        if v is not None
2959                    }
2960                )
2961
2962                propagated_experiment_attributes = PropagatedExperimentAttributes(
2963                    experiment_id=experiment_id,
2964                    experiment_name=experiment_run_name,
2965                    experiment_metadata=_serialize(experiment_metadata),
2966                    experiment_dataset_id=dataset_id,
2967                    experiment_item_id=experiment_item_id,
2968                    experiment_item_metadata=_serialize(item_metadata),
2969                    experiment_item_root_observation_id=span.id,
2970                )
2971
2972                with _propagate_attributes(experiment=propagated_experiment_attributes):
2973                    output = await _run_task(task, item)
2974
2975                span.update(
2976                    input=input_data,
2977                    output=output,
2978                    metadata=final_observation_metadata,
2979                )
2980
2981            except Exception as e:
2982                span.update(
2983                    output=f"Error: {str(e)}", level="ERROR", status_message=str(e)
2984                )
2985                raise e
2986
2987            # Run evaluators
2988            evaluations = []
2989
2990            for evaluator in evaluators:
2991                try:
2992                    eval_metadata: Optional[Dict[str, Any]] = None
2993
2994                    if isinstance(item, dict):
2995                        eval_metadata = item.get("metadata")
2996                    elif hasattr(item, "metadata"):
2997                        eval_metadata = item.metadata
2998
2999                    with _propagate_attributes(
3000                        experiment=propagated_experiment_attributes
3001                    ):
3002                        eval_results = await _run_evaluator(
3003                            evaluator,
3004                            input=input_data,
3005                            output=output,
3006                            expected_output=expected_output,
3007                            metadata=eval_metadata,
3008                        )
3009                        evaluations.extend(eval_results)
3010
3011                        # Store evaluations as scores
3012                        for evaluation in eval_results:
3013                            self.create_score(
3014                                trace_id=trace_id,
3015                                observation_id=span.id,
3016                                name=evaluation.name,
3017                                value=evaluation.value,  # type: ignore
3018                                comment=evaluation.comment,
3019                                metadata=evaluation.metadata,
3020                                config_id=evaluation.config_id,
3021                                data_type=evaluation.data_type,  # type: ignore
3022                            )
3023
3024                except Exception as e:
3025                    langfuse_logger.error(f"Evaluator failed: {e}")
3026
3027            # Run composite evaluator if provided and we have evaluations
3028            if composite_evaluator and evaluations:
3029                try:
3030                    composite_eval_metadata: Optional[Dict[str, Any]] = None
3031                    if isinstance(item, dict):
3032                        composite_eval_metadata = item.get("metadata")
3033                    elif hasattr(item, "metadata"):
3034                        composite_eval_metadata = item.metadata
3035
3036                    with _propagate_attributes(
3037                        experiment=propagated_experiment_attributes
3038                    ):
3039                        result = composite_evaluator(
3040                            input=input_data,
3041                            output=output,
3042                            expected_output=expected_output,
3043                            metadata=composite_eval_metadata,
3044                            evaluations=evaluations,
3045                        )
3046
3047                        # Handle async composite evaluators
3048                        if asyncio.iscoroutine(result):
3049                            result = await result
3050
3051                        # Normalize to list
3052                        composite_evals: List[Evaluation] = []
3053                        if isinstance(result, (dict, Evaluation)):
3054                            composite_evals = [result]  # type: ignore
3055                        elif isinstance(result, list):
3056                            composite_evals = result  # type: ignore
3057
3058                        # Store composite evaluations as scores and add to evaluations list
3059                        for composite_evaluation in composite_evals:
3060                            self.create_score(
3061                                trace_id=trace_id,
3062                                observation_id=span.id,
3063                                name=composite_evaluation.name,
3064                                value=composite_evaluation.value,  # type: ignore
3065                                comment=composite_evaluation.comment,
3066                                metadata=composite_evaluation.metadata,
3067                                config_id=composite_evaluation.config_id,
3068                                data_type=composite_evaluation.data_type,  # type: ignore
3069                            )
3070                            evaluations.append(composite_evaluation)
3071
3072                except Exception as e:
3073                    langfuse_logger.error(f"Composite evaluator failed: {e}")
3074
3075            return ExperimentItemResult(
3076                item=item,
3077                output=output,
3078                evaluations=evaluations,
3079                trace_id=trace_id,
3080                dataset_run_id=dataset_run_id,
3081            )
3082
3083    def _create_experiment_run_name(
3084        self, *, name: Optional[str] = None, run_name: Optional[str] = None
3085    ) -> str:
3086        if run_name:
3087            return run_name
3088
3089        iso_timestamp = _get_timestamp().isoformat().replace("+00:00", "Z")
3090
3091        return f"{name} - {iso_timestamp}"
3092
3093    def run_batched_evaluation(
3094        self,
3095        *,
3096        scope: Literal["traces", "observations"],
3097        mapper: MapperFunction,
3098        filter: Optional[str] = None,
3099        fetch_batch_size: int = 50,
3100        max_items: Optional[int] = None,
3101        max_retries: int = 3,
3102        evaluators: List[EvaluatorFunction],
3103        composite_evaluator: Optional[CompositeEvaluatorFunction] = None,
3104        max_concurrency: int = 50,
3105        metadata: Optional[Dict[str, Any]] = None,
3106        resume_from: Optional[BatchEvaluationResumeToken] = None,
3107        verbose: bool = False,
3108    ) -> BatchEvaluationResult:
3109        """Fetch traces or observations and run evaluations on each item.
3110
3111        This method provides a powerful way to evaluate existing data in Langfuse at scale.
3112        It fetches items based on filters, transforms them using a mapper function, runs
3113        evaluators on each item, and creates scores that are linked back to the original
3114        entities. This is ideal for:
3115
3116        - Running evaluations on production traces after deployment
3117        - Backtesting new evaluation metrics on historical data
3118        - Batch scoring of observations for quality monitoring
3119        - Periodic evaluation runs on recent data
3120
3121        The method uses a streaming/pipeline approach to process items in batches, making
3122        it memory-efficient for large datasets. It includes comprehensive error handling,
3123        retry logic, and resume capability for long-running evaluations.
3124
3125        Args:
3126            scope: The type of items to evaluate. Must be one of:
3127                - "traces": Evaluate complete traces with all their observations
3128                - "observations": Evaluate individual observations (spans, generations, events)
3129            mapper: Function that transforms API response objects into evaluator inputs.
3130                Receives a trace/observation object and returns an EvaluatorInputs
3131                instance with input, output, expected_output, and metadata fields.
3132                Can be sync or async.
3133            evaluators: List of evaluation functions to run on each item. Each evaluator
3134                receives the mapped inputs and returns Evaluation object(s). Evaluator
3135                failures are logged but don't stop the batch evaluation.
3136            filter: Optional JSON filter string for querying items (same format as Langfuse API). Examples:
3137                - '{"tags": ["production"]}'
3138                - '{"user_id": "user123", "timestamp": {"operator": ">", "value": "2024-01-01"}}'
3139                Default: None (fetches all items).
3140            fetch_batch_size: Number of items to fetch per API call and hold in memory.
3141                Larger values may be faster but use more memory. Default: 50.
3142            max_items: Maximum total number of items to process. If None, processes all
3143                items matching the filter. Useful for testing or limiting evaluation runs.
3144                Default: None (process all).
3145            max_concurrency: Maximum number of items to evaluate concurrently. Controls
3146                parallelism and resource usage. Default: 50.
3147            composite_evaluator: Optional function that creates a composite score from
3148                item-level evaluations. Receives the original item and its evaluations,
3149                returns a single Evaluation. Useful for weighted averages or combined metrics.
3150                Default: None.
3151            metadata: Optional metadata dict to add to all created scores. Useful for
3152                tracking evaluation runs, versions, or other context. Default: None.
3153            max_retries: Maximum number of retry attempts for failed batch fetches.
3154                Uses exponential backoff (1s, 2s, 4s). Default: 3.
3155            verbose: If True, logs progress information to console. Useful for monitoring
3156                long-running evaluations. Default: False.
3157            resume_from: Optional resume token from a previous incomplete run. Allows
3158                continuing evaluation after interruption or failure. Default: None.
3159
3160
3161        Returns:
3162            BatchEvaluationResult containing:
3163                - total_items_fetched: Number of items fetched from API
3164                - total_items_processed: Number of items successfully evaluated
3165                - total_items_failed: Number of items that failed evaluation
3166                - total_scores_created: Scores created by item-level evaluators
3167                - total_composite_scores_created: Scores created by composite evaluator
3168                - total_evaluations_failed: Individual evaluator failures
3169                - evaluator_stats: Per-evaluator statistics (success rate, scores created)
3170                - resume_token: Token for resuming if incomplete (None if completed)
3171                - completed: True if all items processed
3172                - duration_seconds: Total execution time
3173                - failed_item_ids: IDs of items that failed
3174                - error_summary: Error types and counts
3175                - has_more_items: True if max_items reached but more exist
3176
3177        Raises:
3178            ValueError: If invalid scope is provided.
3179
3180        Examples:
3181            Basic trace evaluation:
3182            ```python
3183            from langfuse import Langfuse, EvaluatorInputs, Evaluation
3184
3185            client = Langfuse()
3186
3187            # Define mapper to extract fields from traces
3188            def trace_mapper(trace):
3189                return EvaluatorInputs(
3190                    input=trace.input,
3191                    output=trace.output,
3192                    expected_output=None,
3193                    metadata={"trace_id": trace.id}
3194                )
3195
3196            # Define evaluator
3197            def length_evaluator(*, input, output, expected_output, metadata):
3198                return Evaluation(
3199                    name="output_length",
3200                    value=len(output) if output else 0
3201                )
3202
3203            # Run batch evaluation
3204            result = client.run_batched_evaluation(
3205                scope="traces",
3206                mapper=trace_mapper,
3207                evaluators=[length_evaluator],
3208                filter='{"tags": ["production"]}',
3209                max_items=1000,
3210                verbose=True
3211            )
3212
3213            print(f"Processed {result.total_items_processed} traces")
3214            print(f"Created {result.total_scores_created} scores")
3215            ```
3216
3217            Evaluation with composite scorer:
3218            ```python
3219            def accuracy_evaluator(*, input, output, expected_output, metadata):
3220                # ... evaluation logic
3221                return Evaluation(name="accuracy", value=0.85)
3222
3223            def relevance_evaluator(*, input, output, expected_output, metadata):
3224                # ... evaluation logic
3225                return Evaluation(name="relevance", value=0.92)
3226
3227            def composite_evaluator(*, item, evaluations):
3228                # Weighted average of evaluations
3229                weights = {"accuracy": 0.6, "relevance": 0.4}
3230                total = sum(
3231                    e.value * weights.get(e.name, 0)
3232                    for e in evaluations
3233                    if isinstance(e.value, (int, float))
3234                )
3235                return Evaluation(
3236                    name="composite_score",
3237                    value=total,
3238                    comment=f"Weighted average of {len(evaluations)} metrics"
3239                )
3240
3241            result = client.run_batched_evaluation(
3242                scope="traces",
3243                mapper=trace_mapper,
3244                evaluators=[accuracy_evaluator, relevance_evaluator],
3245                composite_evaluator=composite_evaluator,
3246                filter='{"user_id": "important_user"}',
3247                verbose=True
3248            )
3249            ```
3250
3251            Handling incomplete runs with resume:
3252            ```python
3253            # Initial run that may fail or timeout
3254            result = client.run_batched_evaluation(
3255                scope="observations",
3256                mapper=obs_mapper,
3257                evaluators=[my_evaluator],
3258                max_items=10000,
3259                verbose=True
3260            )
3261
3262            # Check if incomplete
3263            if not result.completed and result.resume_token:
3264                print(f"Processed {result.resume_token.items_processed} items before interruption")
3265
3266                # Resume from where it left off
3267                result = client.run_batched_evaluation(
3268                    scope="observations",
3269                    mapper=obs_mapper,
3270                    evaluators=[my_evaluator],
3271                    resume_from=result.resume_token,
3272                    verbose=True
3273                )
3274
3275            print(f"Total items processed: {result.total_items_processed}")
3276            ```
3277
3278            Monitoring evaluator performance:
3279            ```python
3280            result = client.run_batched_evaluation(...)
3281
3282            for stats in result.evaluator_stats:
3283                success_rate = stats.successful_runs / stats.total_runs
3284                print(f"{stats.name}:")
3285                print(f"  Success rate: {success_rate:.1%}")
3286                print(f"  Scores created: {stats.total_scores_created}")
3287
3288                if stats.failed_runs > 0:
3289                    print(f"  ⚠️  Failed {stats.failed_runs} times")
3290            ```
3291
3292        Note:
3293            - Evaluator failures are logged but don't stop the batch evaluation
3294            - Individual item failures are tracked but don't stop processing
3295            - Fetch failures are retried with exponential backoff
3296            - All scores are automatically flushed to Langfuse at the end
3297            - The resume mechanism uses timestamp-based filtering to avoid duplicates
3298        """
3299        runner = BatchEvaluationRunner(self)
3300
3301        return cast(
3302            BatchEvaluationResult,
3303            run_async_safely(
3304                runner.run_async(
3305                    scope=scope,
3306                    mapper=mapper,
3307                    evaluators=evaluators,
3308                    filter=filter,
3309                    fetch_batch_size=fetch_batch_size,
3310                    max_items=max_items,
3311                    max_concurrency=max_concurrency,
3312                    composite_evaluator=composite_evaluator,
3313                    metadata=metadata,
3314                    max_retries=max_retries,
3315                    verbose=verbose,
3316                    resume_from=resume_from,
3317                )
3318            ),
3319        )
3320
3321    def auth_check(self) -> bool:
3322        """Check if the provided credentials (public and secret key) are valid.
3323
3324        Raises:
3325            Exception: If no projects were found for the provided credentials.
3326
3327        Note:
3328            This method is blocking. It is discouraged to use it in production code.
3329        """
3330        try:
3331            projects = self.api.projects.get()
3332            langfuse_logger.debug(
3333                f"Auth check successful, found {len(projects.data)} projects"
3334            )
3335            if len(projects.data) == 0:
3336                raise Exception(
3337                    "Auth check failed, no project found for the keys provided."
3338                )
3339            return True
3340
3341        except AttributeError as e:
3342            langfuse_logger.warning(
3343                f"Auth check failed: Client not properly initialized. Error: {e}"
3344            )
3345            return False
3346
3347        except Error as e:
3348            handle_fern_exception(e)
3349            raise e
3350
3351    def create_dataset(
3352        self,
3353        *,
3354        name: str,
3355        description: Optional[str] = None,
3356        metadata: Optional[Any] = None,
3357        input_schema: Optional[Any] = None,
3358        expected_output_schema: Optional[Any] = None,
3359    ) -> Dataset:
3360        """Create a dataset with the given name on Langfuse.
3361
3362        Args:
3363            name: Name of the dataset to create.
3364            description: Description of the dataset. Defaults to None.
3365            metadata: Additional metadata. Defaults to None.
3366            input_schema: JSON Schema for validating dataset item inputs. When set, all new items will be validated against this schema.
3367            expected_output_schema: JSON Schema for validating dataset item expected outputs. When set, all new items will be validated against this schema.
3368
3369        Returns:
3370            Dataset: The created dataset as returned by the Langfuse API.
3371        """
3372        try:
3373            body = CreateDatasetRequest(
3374                name=name,
3375                description=description,
3376                metadata=metadata,
3377                inputSchema=input_schema,
3378                expectedOutputSchema=expected_output_schema,
3379            )
3380            langfuse_logger.debug(f"Creating datasets {body}")
3381
3382            return self.api.datasets.create(request=body)
3383
3384        except Error as e:
3385            handle_fern_exception(e)
3386            raise e
3387
3388    def create_dataset_item(
3389        self,
3390        *,
3391        dataset_name: str,
3392        input: Optional[Any] = None,
3393        expected_output: Optional[Any] = None,
3394        metadata: Optional[Any] = None,
3395        source_trace_id: Optional[str] = None,
3396        source_observation_id: Optional[str] = None,
3397        status: Optional[DatasetStatus] = None,
3398        id: Optional[str] = None,
3399    ) -> DatasetItem:
3400        """Create a dataset item.
3401
3402        Upserts if an item with id already exists.
3403
3404        Args:
3405            dataset_name: Name of the dataset in which the dataset item should be created.
3406            input: Input data. Defaults to None. Can contain any dict, list or scalar.
3407            expected_output: Expected output data. Defaults to None. Can contain any dict, list or scalar.
3408            metadata: Additional metadata. Defaults to None. Can contain any dict, list or scalar.
3409            source_trace_id: Id of the source trace. Defaults to None.
3410            source_observation_id: Id of the source observation. Defaults to None.
3411            status: Status of the dataset item. Defaults to ACTIVE for newly created items.
3412            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.
3413
3414        Returns:
3415            DatasetItem: The created dataset item as returned by the Langfuse API.
3416
3417        Example:
3418            ```python
3419            from langfuse import Langfuse
3420
3421            langfuse = Langfuse()
3422
3423            # Uploading items to the Langfuse dataset named "capital_cities"
3424            langfuse.create_dataset_item(
3425                dataset_name="capital_cities",
3426                input={"input": {"country": "Italy"}},
3427                expected_output={"expected_output": "Rome"},
3428                metadata={"foo": "bar"}
3429            )
3430            ```
3431        """
3432        try:
3433            body = CreateDatasetItemRequest(
3434                datasetName=dataset_name,
3435                input=input,
3436                expectedOutput=expected_output,
3437                metadata=metadata,
3438                sourceTraceId=source_trace_id,
3439                sourceObservationId=source_observation_id,
3440                status=status,
3441                id=id,
3442            )
3443            langfuse_logger.debug(f"Creating dataset item {body}")
3444            return self.api.dataset_items.create(request=body)
3445        except Error as e:
3446            handle_fern_exception(e)
3447            raise e
3448
3449    def resolve_media_references(
3450        self,
3451        *,
3452        obj: Any,
3453        resolve_with: Literal["base64_data_uri"],
3454        max_depth: int = 10,
3455        content_fetch_timeout_seconds: int = 5,
3456    ) -> Any:
3457        """Replace media reference strings in an object with base64 data URIs.
3458
3459        This method recursively traverses an object (up to max_depth) looking for media reference strings
3460        in the format "@@@langfuseMedia:...@@@". When found, it (synchronously) fetches the actual media content using
3461        the provided Langfuse client and replaces the reference string with a base64 data URI.
3462
3463        If fetching media content fails for a reference string, a warning is logged and the reference
3464        string is left unchanged.
3465
3466        Args:
3467            obj: The object to process. Can be a primitive value, array, or nested object.
3468                If the object has a __dict__ attribute, a dict will be returned instead of the original object type.
3469            resolve_with: The representation of the media content to replace the media reference string with.
3470                Currently only "base64_data_uri" is supported.
3471            max_depth: int: The maximum depth to traverse the object. Default is 10.
3472            content_fetch_timeout_seconds: int: The timeout in seconds for fetching media content. Default is 5.
3473
3474        Returns:
3475            A deep copy of the input object with all media references replaced with base64 data URIs where possible.
3476            If the input object has a __dict__ attribute, a dict will be returned instead of the original object type.
3477
3478        Example:
3479            obj = {
3480                "image": "@@@langfuseMedia:type=image/jpeg|id=123|source=bytes@@@",
3481                "nested": {
3482                    "pdf": "@@@langfuseMedia:type=application/pdf|id=456|source=bytes@@@"
3483                }
3484            }
3485
3486            result = await LangfuseMedia.resolve_media_references(obj, langfuse_client)
3487
3488            # Result:
3489            # {
3490            #     "image": "data:image/jpeg;base64,/9j/4AAQSkZJRg...",
3491            #     "nested": {
3492            #         "pdf": "data:application/pdf;base64,JVBERi0xLjcK..."
3493            #     }
3494            # }
3495        """
3496        return LangfuseMedia.resolve_media_references(
3497            langfuse_client=self,
3498            obj=obj,
3499            resolve_with=resolve_with,
3500            max_depth=max_depth,
3501            content_fetch_timeout_seconds=content_fetch_timeout_seconds,
3502        )
3503
3504    @overload
3505    def get_prompt(
3506        self,
3507        name: str,
3508        *,
3509        version: Optional[int] = None,
3510        label: Optional[str] = None,
3511        type: Literal["chat"],
3512        cache_ttl_seconds: Optional[int] = None,
3513        fallback: Optional[List[ChatMessageDict]] = None,
3514        max_retries: Optional[int] = None,
3515        fetch_timeout_seconds: Optional[int] = None,
3516    ) -> ChatPromptClient: ...
3517
3518    @overload
3519    def get_prompt(
3520        self,
3521        name: str,
3522        *,
3523        version: Optional[int] = None,
3524        label: Optional[str] = None,
3525        type: Literal["text"] = "text",
3526        cache_ttl_seconds: Optional[int] = None,
3527        fallback: Optional[str] = None,
3528        max_retries: Optional[int] = None,
3529        fetch_timeout_seconds: Optional[int] = None,
3530    ) -> TextPromptClient: ...
3531
3532    def get_prompt(
3533        self,
3534        name: str,
3535        *,
3536        version: Optional[int] = None,
3537        label: Optional[str] = None,
3538        type: Literal["chat", "text"] = "text",
3539        cache_ttl_seconds: Optional[int] = None,
3540        fallback: Union[Optional[List[ChatMessageDict]], Optional[str]] = None,
3541        max_retries: Optional[int] = None,
3542        fetch_timeout_seconds: Optional[int] = None,
3543    ) -> PromptClient:
3544        """Get a prompt.
3545
3546        This method attempts to fetch the requested prompt from the local cache. If the prompt is not found
3547        in the cache or if the cached prompt has expired, it will try to fetch the prompt from the server again
3548        and update the cache. If fetching the new prompt fails, and there is an expired prompt in the cache, it will
3549        return the expired prompt as a fallback.
3550
3551        Args:
3552            name (str): The name of the prompt to retrieve.
3553
3554        Keyword Args:
3555            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.
3556            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.
3557            cache_ttl_seconds: Optional[int]: Time-to-live in seconds for caching the prompt. Must be specified as a
3558            keyword argument. If not set, defaults to 60 seconds. Disables caching if set to 0.
3559            type: Literal["chat", "text"]: The type of the prompt to retrieve. Defaults to "text".
3560            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.
3561            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.
3562            fetch_timeout_seconds: Optional[int]: The timeout in milliseconds for fetching the prompt. Defaults to the default timeout set on the SDK, which is 5 seconds per default.
3563
3564        Returns:
3565            The prompt object retrieved from the cache or directly fetched if not cached or expired of type
3566            - TextPromptClient, if type argument is 'text'.
3567            - ChatPromptClient, if type argument is 'chat'.
3568
3569        Raises:
3570            Exception: Propagates any exceptions raised during the fetching of a new prompt, unless there is an
3571            expired prompt in the cache, in which case it logs a warning and returns the expired prompt.
3572        """
3573        if self._resources is None:
3574            raise Error(
3575                "SDK is not correctly initialized. Check the init logs for more details."
3576            )
3577        if version is not None and label is not None:
3578            raise ValueError("Cannot specify both version and label at the same time.")
3579
3580        if not name:
3581            raise ValueError("Prompt name cannot be empty.")
3582
3583        cache_key = PromptCache.generate_cache_key(name, version=version, label=label)
3584        bounded_max_retries = self._get_bounded_max_retries(
3585            max_retries, default_max_retries=2, max_retries_upper_bound=4
3586        )
3587
3588        langfuse_logger.debug(f"Getting prompt '{cache_key}'")
3589        cached_prompt = self._resources.prompt_cache.get(cache_key)
3590
3591        if cached_prompt is None or cache_ttl_seconds == 0:
3592            langfuse_logger.debug(
3593                f"Prompt '{cache_key}' not found in cache or caching disabled."
3594            )
3595            try:
3596                return self._fetch_prompt_and_update_cache(
3597                    name,
3598                    version=version,
3599                    label=label,
3600                    ttl_seconds=cache_ttl_seconds,
3601                    max_retries=bounded_max_retries,
3602                    fetch_timeout_seconds=fetch_timeout_seconds,
3603                )
3604            except Exception as e:
3605                if fallback:
3606                    langfuse_logger.warning(
3607                        f"Returning fallback prompt for '{cache_key}' due to fetch error: {e}"
3608                    )
3609
3610                    fallback_client_args: Dict[str, Any] = {
3611                        "name": name,
3612                        "prompt": fallback,
3613                        "type": type,
3614                        "version": version or 0,
3615                        "config": {},
3616                        "labels": [label] if label else [],
3617                        "tags": [],
3618                    }
3619
3620                    if type == "text":
3621                        return TextPromptClient(
3622                            prompt=Prompt_Text(**fallback_client_args),
3623                            is_fallback=True,
3624                        )
3625
3626                    if type == "chat":
3627                        return ChatPromptClient(
3628                            prompt=Prompt_Chat(**fallback_client_args),
3629                            is_fallback=True,
3630                        )
3631
3632                raise e
3633
3634        if cached_prompt.is_expired():
3635            langfuse_logger.debug(f"Stale prompt '{cache_key}' found in cache.")
3636            try:
3637                # refresh prompt in background thread, refresh_prompt deduplicates tasks
3638                langfuse_logger.debug(f"Refreshing prompt '{cache_key}' in background.")
3639
3640                def refresh_task() -> None:
3641                    self._fetch_prompt_and_update_cache(
3642                        name,
3643                        version=version,
3644                        label=label,
3645                        ttl_seconds=cache_ttl_seconds,
3646                        max_retries=bounded_max_retries,
3647                        fetch_timeout_seconds=fetch_timeout_seconds,
3648                    )
3649
3650                self._resources.prompt_cache.add_refresh_prompt_task(
3651                    cache_key,
3652                    refresh_task,
3653                )
3654                langfuse_logger.debug(
3655                    f"Returning stale prompt '{cache_key}' from cache."
3656                )
3657                # return stale prompt
3658                return cached_prompt.value
3659
3660            except Exception as e:
3661                langfuse_logger.warning(
3662                    f"Error when refreshing cached prompt '{cache_key}', returning cached version. Error: {e}"
3663                )
3664                # creation of refresh prompt task failed, return stale prompt
3665                return cached_prompt.value
3666
3667        return cached_prompt.value
3668
3669    def _fetch_prompt_and_update_cache(
3670        self,
3671        name: str,
3672        *,
3673        version: Optional[int] = None,
3674        label: Optional[str] = None,
3675        ttl_seconds: Optional[int] = None,
3676        max_retries: int,
3677        fetch_timeout_seconds: Optional[int],
3678    ) -> PromptClient:
3679        cache_key = PromptCache.generate_cache_key(name, version=version, label=label)
3680        langfuse_logger.debug(f"Fetching prompt '{cache_key}' from server...")
3681
3682        try:
3683
3684            @backoff.on_exception(
3685                backoff.constant, Exception, max_tries=max_retries + 1, logger=None
3686            )
3687            def fetch_prompts() -> Any:
3688                return self.api.prompts.get(
3689                    self._url_encode(name),
3690                    version=version,
3691                    label=label,
3692                    request_options={
3693                        "timeout_in_seconds": fetch_timeout_seconds,
3694                    }
3695                    if fetch_timeout_seconds is not None
3696                    else None,
3697                )
3698
3699            prompt_response = fetch_prompts()
3700
3701            prompt: PromptClient
3702            if prompt_response.type == "chat":
3703                prompt = ChatPromptClient(prompt_response)
3704            else:
3705                prompt = TextPromptClient(prompt_response)
3706
3707            if self._resources is not None:
3708                self._resources.prompt_cache.set(cache_key, prompt, ttl_seconds)
3709
3710            return prompt
3711
3712        except NotFoundError as not_found_error:
3713            langfuse_logger.warning(
3714                f"Prompt '{cache_key}' not found during refresh, evicting from cache."
3715            )
3716            if self._resources is not None:
3717                self._resources.prompt_cache.delete(cache_key)
3718            raise not_found_error
3719
3720        except Exception as e:
3721            langfuse_logger.error(
3722                f"Error while fetching prompt '{cache_key}': {str(e)}"
3723            )
3724            raise e
3725
3726    def _get_bounded_max_retries(
3727        self,
3728        max_retries: Optional[int],
3729        *,
3730        default_max_retries: int = 2,
3731        max_retries_upper_bound: int = 4,
3732    ) -> int:
3733        if max_retries is None:
3734            return default_max_retries
3735
3736        bounded_max_retries = min(
3737            max(max_retries, 0),
3738            max_retries_upper_bound,
3739        )
3740
3741        return bounded_max_retries
3742
3743    @overload
3744    def create_prompt(
3745        self,
3746        *,
3747        name: str,
3748        prompt: List[Union[ChatMessageDict, ChatMessageWithPlaceholdersDict]],
3749        labels: List[str] = [],
3750        tags: Optional[List[str]] = None,
3751        type: Optional[Literal["chat"]],
3752        config: Optional[Any] = None,
3753        commit_message: Optional[str] = None,
3754    ) -> ChatPromptClient: ...
3755
3756    @overload
3757    def create_prompt(
3758        self,
3759        *,
3760        name: str,
3761        prompt: str,
3762        labels: List[str] = [],
3763        tags: Optional[List[str]] = None,
3764        type: Optional[Literal["text"]] = "text",
3765        config: Optional[Any] = None,
3766        commit_message: Optional[str] = None,
3767    ) -> TextPromptClient: ...
3768
3769    def create_prompt(
3770        self,
3771        *,
3772        name: str,
3773        prompt: Union[
3774            str, List[Union[ChatMessageDict, ChatMessageWithPlaceholdersDict]]
3775        ],
3776        labels: List[str] = [],
3777        tags: Optional[List[str]] = None,
3778        type: Optional[Literal["chat", "text"]] = "text",
3779        config: Optional[Any] = None,
3780        commit_message: Optional[str] = None,
3781    ) -> PromptClient:
3782        """Create a new prompt in Langfuse.
3783
3784        Keyword Args:
3785            name : The name of the prompt to be created.
3786            prompt : The content of the prompt to be created.
3787            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.
3788            labels: The labels of the prompt. Defaults to None. To create a default-served prompt, add the 'production' label.
3789            tags: The tags of the prompt. Defaults to None. Will be applied to all versions of the prompt.
3790            config: Additional structured data to be saved with the prompt. Defaults to None.
3791            type: The type of the prompt to be created. "chat" vs. "text". Defaults to "text".
3792            commit_message: Optional string describing the change.
3793
3794        Returns:
3795            TextPromptClient: The prompt if type argument is 'text'.
3796            ChatPromptClient: The prompt if type argument is 'chat'.
3797        """
3798        try:
3799            langfuse_logger.debug(f"Creating prompt {name=}, {labels=}")
3800
3801            if type == "chat":
3802                if not isinstance(prompt, list):
3803                    raise ValueError(
3804                        "For 'chat' type, 'prompt' must be a list of chat messages with role and content attributes."
3805                    )
3806                request: Union[CreatePromptRequest_Chat, CreatePromptRequest_Text] = (
3807                    CreatePromptRequest_Chat(
3808                        name=name,
3809                        prompt=cast(Any, prompt),
3810                        labels=labels,
3811                        tags=tags,
3812                        config=config or {},
3813                        commitMessage=commit_message,
3814                        type="chat",
3815                    )
3816                )
3817                server_prompt = self.api.prompts.create(request=request)
3818
3819                if self._resources is not None:
3820                    self._resources.prompt_cache.invalidate(name)
3821
3822                return ChatPromptClient(prompt=cast(Prompt_Chat, server_prompt))
3823
3824            if not isinstance(prompt, str):
3825                raise ValueError("For 'text' type, 'prompt' must be a string.")
3826
3827            request = CreatePromptRequest_Text(
3828                name=name,
3829                prompt=prompt,
3830                labels=labels,
3831                tags=tags,
3832                config=config or {},
3833                commitMessage=commit_message,
3834                type="text",
3835            )
3836
3837            server_prompt = self.api.prompts.create(request=request)
3838
3839            if self._resources is not None:
3840                self._resources.prompt_cache.invalidate(name)
3841
3842            return TextPromptClient(prompt=cast(Prompt_Text, server_prompt))
3843
3844        except Error as e:
3845            handle_fern_exception(e)
3846            raise e
3847
3848    def update_prompt(
3849        self,
3850        *,
3851        name: str,
3852        version: int,
3853        new_labels: List[str] = [],
3854    ) -> Any:
3855        """Update an existing prompt version in Langfuse. The Langfuse SDK prompt cache is invalidated for all prompts witht he specified name.
3856
3857        Args:
3858            name (str): The name of the prompt to update.
3859            version (int): The version number of the prompt to update.
3860            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 [].
3861
3862        Returns:
3863            Prompt: The updated prompt from the Langfuse API.
3864
3865        """
3866        updated_prompt = self.api.prompt_version.update(
3867            name=self._url_encode(name),
3868            version=version,
3869            new_labels=new_labels,
3870        )
3871
3872        if self._resources is not None:
3873            self._resources.prompt_cache.invalidate(name)
3874
3875        return updated_prompt
3876
3877    def _url_encode(self, url: str, *, is_url_param: Optional[bool] = False) -> str:
3878        # httpx ≥ 0.28 does its own WHATWG-compliant quoting (eg. encodes bare
3879        # “%”, “?”, “#”, “|”, … in query/path parts).  Re-quoting here would
3880        # double-encode, so we skip when the value is about to be sent straight
3881        # to httpx (`is_url_param=True`) and the installed version is ≥ 0.28.
3882        if is_url_param and Version(httpx.__version__) >= Version("0.28.0"):
3883            return url
3884
3885        # urllib.parse.quote does not escape slashes "/" by default; we need to add safe="" to force escaping
3886        # we need add safe="" to force escaping of slashes
3887        # This is necessary for prompts in prompt folders
3888        return urllib.parse.quote(url, safe="")
3889
3890    def clear_prompt_cache(self) -> None:
3891        """Clear the entire prompt cache, removing all cached prompts.
3892
3893        This method is useful when you want to force a complete refresh of all
3894        cached prompts, for example after major updates or when you need to
3895        ensure the latest versions are fetched from the server.
3896        """
3897        if self._resources is not None:
3898            self._resources.prompt_cache.clear()