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    ) -> None: ...
2113
2114    @overload
2115    def score_current_span(
2116        self,
2117        *,
2118        name: str,
2119        value: str,
2120        score_id: Optional[str] = None,
2121        data_type: Optional[Literal["CATEGORICAL"]] = "CATEGORICAL",
2122        comment: Optional[str] = None,
2123        config_id: Optional[str] = None,
2124    ) -> None: ...
2125
2126    def score_current_span(
2127        self,
2128        *,
2129        name: str,
2130        value: Union[float, str],
2131        score_id: Optional[str] = None,
2132        data_type: Optional[ScoreDataType] = None,
2133        comment: Optional[str] = None,
2134        config_id: Optional[str] = None,
2135    ) -> None:
2136        """Create a score for the current active span.
2137
2138        This method scores the currently active span in the context. It's a convenient
2139        way to score the current operation without needing to know its trace and span IDs.
2140
2141        Args:
2142            name: Name of the score (e.g., "relevance", "accuracy")
2143            value: Score value (can be numeric for NUMERIC/BOOLEAN types or string for CATEGORICAL)
2144            score_id: Optional custom ID for the score (auto-generated if not provided)
2145            data_type: Type of score (NUMERIC, BOOLEAN, or CATEGORICAL)
2146            comment: Optional comment or explanation for the score
2147            config_id: Optional ID of a score config defined in Langfuse
2148
2149        Example:
2150            ```python
2151            with langfuse.start_as_current_generation(name="answer-query") as generation:
2152                # Generate answer
2153                response = generate_answer(...)
2154                generation.update(output=response)
2155
2156                # Score the generation
2157                langfuse.score_current_span(
2158                    name="relevance",
2159                    value=0.85,
2160                    data_type="NUMERIC",
2161                    comment="Mostly relevant but contains some tangential information"
2162                )
2163            ```
2164        """
2165        current_span = self._get_current_otel_span()
2166
2167        if current_span is not None:
2168            trace_id = self._get_otel_trace_id(current_span)
2169            observation_id = self._get_otel_span_id(current_span)
2170
2171            langfuse_logger.info(
2172                f"Score: Creating score name='{name}' value={value} for current span ({observation_id}) in trace {trace_id}"
2173            )
2174
2175            self.create_score(
2176                trace_id=trace_id,
2177                observation_id=observation_id,
2178                name=name,
2179                value=cast(str, value),
2180                score_id=score_id,
2181                data_type=cast(Literal["CATEGORICAL"], data_type),
2182                comment=comment,
2183                config_id=config_id,
2184            )
2185
2186    @overload
2187    def score_current_trace(
2188        self,
2189        *,
2190        name: str,
2191        value: float,
2192        score_id: Optional[str] = None,
2193        data_type: Optional[Literal["NUMERIC", "BOOLEAN"]] = None,
2194        comment: Optional[str] = None,
2195        config_id: Optional[str] = None,
2196    ) -> None: ...
2197
2198    @overload
2199    def score_current_trace(
2200        self,
2201        *,
2202        name: str,
2203        value: str,
2204        score_id: Optional[str] = None,
2205        data_type: Optional[Literal["CATEGORICAL"]] = "CATEGORICAL",
2206        comment: Optional[str] = None,
2207        config_id: Optional[str] = None,
2208    ) -> None: ...
2209
2210    def score_current_trace(
2211        self,
2212        *,
2213        name: str,
2214        value: Union[float, str],
2215        score_id: Optional[str] = None,
2216        data_type: Optional[ScoreDataType] = None,
2217        comment: Optional[str] = None,
2218        config_id: Optional[str] = None,
2219    ) -> None:
2220        """Create a score for the current trace.
2221
2222        This method scores the trace of the currently active span. Unlike score_current_span,
2223        this method associates the score with the entire trace rather than a specific span.
2224        It's useful for scoring overall performance or quality of the entire operation.
2225
2226        Args:
2227            name: Name of the score (e.g., "user_satisfaction", "overall_quality")
2228            value: Score value (can be numeric for NUMERIC/BOOLEAN types or string for CATEGORICAL)
2229            score_id: Optional custom ID for the score (auto-generated if not provided)
2230            data_type: Type of score (NUMERIC, BOOLEAN, or CATEGORICAL)
2231            comment: Optional comment or explanation for the score
2232            config_id: Optional ID of a score config defined in Langfuse
2233
2234        Example:
2235            ```python
2236            with langfuse.start_as_current_span(name="process-user-request") as span:
2237                # Process request
2238                result = process_complete_request()
2239                span.update(output=result)
2240
2241                # Score the overall trace
2242                langfuse.score_current_trace(
2243                    name="overall_quality",
2244                    value=0.95,
2245                    data_type="NUMERIC",
2246                    comment="High quality end-to-end response"
2247                )
2248            ```
2249        """
2250        current_span = self._get_current_otel_span()
2251
2252        if current_span is not None:
2253            trace_id = self._get_otel_trace_id(current_span)
2254
2255            langfuse_logger.info(
2256                f"Score: Creating score name='{name}' value={value} for entire trace {trace_id}"
2257            )
2258
2259            self.create_score(
2260                trace_id=trace_id,
2261                name=name,
2262                value=cast(str, value),
2263                score_id=score_id,
2264                data_type=cast(Literal["CATEGORICAL"], data_type),
2265                comment=comment,
2266                config_id=config_id,
2267            )
2268
2269    def flush(self) -> None:
2270        """Force flush all pending spans and events to the Langfuse API.
2271
2272        This method manually flushes any pending spans, scores, and other events to the
2273        Langfuse API. It's useful in scenarios where you want to ensure all data is sent
2274        before proceeding, without waiting for the automatic flush interval.
2275
2276        Example:
2277            ```python
2278            # Record some spans and scores
2279            with langfuse.start_as_current_span(name="operation") as span:
2280                # Do work...
2281                pass
2282
2283            # Ensure all data is sent to Langfuse before proceeding
2284            langfuse.flush()
2285
2286            # Continue with other work
2287            ```
2288        """
2289        if self._resources is not None:
2290            self._resources.flush()
2291
2292    def shutdown(self) -> None:
2293        """Shut down the Langfuse client and flush all pending data.
2294
2295        This method cleanly shuts down the Langfuse client, ensuring all pending data
2296        is flushed to the API and all background threads are properly terminated.
2297
2298        It's important to call this method when your application is shutting down to
2299        prevent data loss and resource leaks. For most applications, using the client
2300        as a context manager or relying on the automatic shutdown via atexit is sufficient.
2301
2302        Example:
2303            ```python
2304            # Initialize Langfuse
2305            langfuse = Langfuse(public_key="...", secret_key="...")
2306
2307            # Use Langfuse throughout your application
2308            # ...
2309
2310            # When application is shutting down
2311            langfuse.shutdown()
2312            ```
2313        """
2314        if self._resources is not None:
2315            self._resources.shutdown()
2316
2317    def get_current_trace_id(self) -> Optional[str]:
2318        """Get the trace ID of the current active span.
2319
2320        This method retrieves the trace ID from the currently active span in the context.
2321        It can be used to get the trace ID for referencing in logs, external systems,
2322        or for creating related operations.
2323
2324        Returns:
2325            The current trace ID as a 32-character lowercase hexadecimal string,
2326            or None if there is no active span.
2327
2328        Example:
2329            ```python
2330            with langfuse.start_as_current_span(name="process-request") as span:
2331                # Get the current trace ID for reference
2332                trace_id = langfuse.get_current_trace_id()
2333
2334                # Use it for external correlation
2335                log.info(f"Processing request with trace_id: {trace_id}")
2336
2337                # Or pass to another system
2338                external_system.process(data, trace_id=trace_id)
2339            ```
2340        """
2341        if not self._tracing_enabled:
2342            langfuse_logger.debug(
2343                "Operation skipped: get_current_trace_id - Tracing is disabled or client is in no-op mode."
2344            )
2345            return None
2346
2347        current_otel_span = self._get_current_otel_span()
2348
2349        return self._get_otel_trace_id(current_otel_span) if current_otel_span else None
2350
2351    def get_current_observation_id(self) -> Optional[str]:
2352        """Get the observation ID (span ID) of the current active span.
2353
2354        This method retrieves the observation ID from the currently active span in the context.
2355        It can be used to get the observation ID for referencing in logs, external systems,
2356        or for creating scores or other related operations.
2357
2358        Returns:
2359            The current observation ID as a 16-character lowercase hexadecimal string,
2360            or None if there is no active span.
2361
2362        Example:
2363            ```python
2364            with langfuse.start_as_current_span(name="process-user-query") as span:
2365                # Get the current observation ID
2366                observation_id = langfuse.get_current_observation_id()
2367
2368                # Store it for later reference
2369                cache.set(f"query_{query_id}_observation", observation_id)
2370
2371                # Process the query...
2372            ```
2373        """
2374        if not self._tracing_enabled:
2375            langfuse_logger.debug(
2376                "Operation skipped: get_current_observation_id - Tracing is disabled or client is in no-op mode."
2377            )
2378            return None
2379
2380        current_otel_span = self._get_current_otel_span()
2381
2382        return self._get_otel_span_id(current_otel_span) if current_otel_span else None
2383
2384    def _get_project_id(self) -> Optional[str]:
2385        """Fetch and return the current project id. Persisted across requests. Returns None if no project id is found for api keys."""
2386        if not self._project_id:
2387            proj = self.api.projects.get()
2388            if not proj.data or not proj.data[0].id:
2389                return None
2390
2391            self._project_id = proj.data[0].id
2392
2393        return self._project_id
2394
2395    def get_trace_url(self, *, trace_id: Optional[str] = None) -> Optional[str]:
2396        """Get the URL to view a trace in the Langfuse UI.
2397
2398        This method generates a URL that links directly to a trace in the Langfuse UI.
2399        It's useful for providing links in logs, notifications, or debugging tools.
2400
2401        Args:
2402            trace_id: Optional trace ID to generate a URL for. If not provided,
2403                     the trace ID of the current active span will be used.
2404
2405        Returns:
2406            A URL string pointing to the trace in the Langfuse UI,
2407            or None if the project ID couldn't be retrieved or no trace ID is available.
2408
2409        Example:
2410            ```python
2411            # Get URL for the current trace
2412            with langfuse.start_as_current_span(name="process-request") as span:
2413                trace_url = langfuse.get_trace_url()
2414                log.info(f"Processing trace: {trace_url}")
2415
2416            # Get URL for a specific trace
2417            specific_trace_url = langfuse.get_trace_url(trace_id="1234567890abcdef1234567890abcdef")
2418            send_notification(f"Review needed for trace: {specific_trace_url}")
2419            ```
2420        """
2421        project_id = self._get_project_id()
2422        final_trace_id = trace_id or self.get_current_trace_id()
2423
2424        return (
2425            f"{self._base_url}/project/{project_id}/traces/{final_trace_id}"
2426            if project_id and final_trace_id
2427            else None
2428        )
2429
2430    def get_dataset(
2431        self, name: str, *, fetch_items_page_size: Optional[int] = 50
2432    ) -> "DatasetClient":
2433        """Fetch a dataset by its name.
2434
2435        Args:
2436            name (str): The name of the dataset to fetch.
2437            fetch_items_page_size (Optional[int]): All items of the dataset will be fetched in chunks of this size. Defaults to 50.
2438
2439        Returns:
2440            DatasetClient: The dataset with the given name.
2441        """
2442        try:
2443            langfuse_logger.debug(f"Getting datasets {name}")
2444            dataset = self.api.datasets.get(dataset_name=self._url_encode(name))
2445
2446            dataset_items = []
2447            page = 1
2448
2449            while True:
2450                new_items = self.api.dataset_items.list(
2451                    dataset_name=self._url_encode(name, is_url_param=True),
2452                    page=page,
2453                    limit=fetch_items_page_size,
2454                )
2455                dataset_items.extend(new_items.data)
2456
2457                if new_items.meta.total_pages <= page:
2458                    break
2459
2460                page += 1
2461
2462            items = [DatasetItemClient(i, langfuse=self) for i in dataset_items]
2463
2464            return DatasetClient(dataset, items=items)
2465
2466        except Error as e:
2467            handle_fern_exception(e)
2468            raise e
2469
2470    def get_dataset_run(
2471        self, *, dataset_name: str, run_name: str
2472    ) -> DatasetRunWithItems:
2473        """Fetch a dataset run by dataset name and run name.
2474
2475        Args:
2476            dataset_name (str): The name of the dataset.
2477            run_name (str): The name of the run.
2478
2479        Returns:
2480            DatasetRunWithItems: The dataset run with its items.
2481        """
2482        try:
2483            return self.api.datasets.get_run(
2484                dataset_name=self._url_encode(dataset_name),
2485                run_name=self._url_encode(run_name),
2486                request_options=None,
2487            )
2488        except Error as e:
2489            handle_fern_exception(e)
2490            raise e
2491
2492    def get_dataset_runs(
2493        self,
2494        *,
2495        dataset_name: str,
2496        page: Optional[int] = None,
2497        limit: Optional[int] = None,
2498    ) -> PaginatedDatasetRuns:
2499        """Fetch all runs for a dataset.
2500
2501        Args:
2502            dataset_name (str): The name of the dataset.
2503            page (Optional[int]): Page number, starts at 1.
2504            limit (Optional[int]): Limit of items per page.
2505
2506        Returns:
2507            PaginatedDatasetRuns: Paginated list of dataset runs.
2508        """
2509        try:
2510            return self.api.datasets.get_runs(
2511                dataset_name=self._url_encode(dataset_name),
2512                page=page,
2513                limit=limit,
2514                request_options=None,
2515            )
2516        except Error as e:
2517            handle_fern_exception(e)
2518            raise e
2519
2520    def delete_dataset_run(
2521        self, *, dataset_name: str, run_name: str
2522    ) -> DeleteDatasetRunResponse:
2523        """Delete a dataset run and all its run items. This action is irreversible.
2524
2525        Args:
2526            dataset_name (str): The name of the dataset.
2527            run_name (str): The name of the run.
2528
2529        Returns:
2530            DeleteDatasetRunResponse: Confirmation of deletion.
2531        """
2532        try:
2533            return self.api.datasets.delete_run(
2534                dataset_name=self._url_encode(dataset_name),
2535                run_name=self._url_encode(run_name),
2536                request_options=None,
2537            )
2538        except Error as e:
2539            handle_fern_exception(e)
2540            raise e
2541
2542    def run_experiment(
2543        self,
2544        *,
2545        name: str,
2546        run_name: Optional[str] = None,
2547        description: Optional[str] = None,
2548        data: ExperimentData,
2549        task: TaskFunction,
2550        evaluators: List[EvaluatorFunction] = [],
2551        composite_evaluator: Optional[CompositeEvaluatorFunction] = None,
2552        run_evaluators: List[RunEvaluatorFunction] = [],
2553        max_concurrency: int = 50,
2554        metadata: Optional[Dict[str, str]] = None,
2555    ) -> ExperimentResult:
2556        """Run an experiment on a dataset with automatic tracing and evaluation.
2557
2558        This method executes a task function on each item in the provided dataset,
2559        automatically traces all executions with Langfuse for observability, runs
2560        item-level and run-level evaluators on the outputs, and returns comprehensive
2561        results with evaluation metrics.
2562
2563        The experiment system provides:
2564        - Automatic tracing of all task executions
2565        - Concurrent processing with configurable limits
2566        - Comprehensive error handling that isolates failures
2567        - Integration with Langfuse datasets for experiment tracking
2568        - Flexible evaluation framework supporting both sync and async evaluators
2569
2570        Args:
2571            name: Human-readable name for the experiment. Used for identification
2572                in the Langfuse UI.
2573            run_name: Optional exact name for the experiment run. If provided, this will be
2574                used as the exact dataset run name if the `data` contains Langfuse dataset items.
2575                If not provided, this will default to the experiment name appended with an ISO timestamp.
2576            description: Optional description explaining the experiment's purpose,
2577                methodology, or expected outcomes.
2578            data: Array of data items to process. Can be either:
2579                - List of dict-like items with 'input', 'expected_output', 'metadata' keys
2580                - List of Langfuse DatasetItem objects from dataset.items
2581            task: Function that processes each data item and returns output.
2582                Must accept 'item' as keyword argument and can return sync or async results.
2583                The task function signature should be: task(*, item, **kwargs) -> Any
2584            evaluators: List of functions to evaluate each item's output individually.
2585                Each evaluator receives input, output, expected_output, and metadata.
2586                Can return single Evaluation dict or list of Evaluation dicts.
2587            composite_evaluator: Optional function that creates composite scores from item-level evaluations.
2588                Receives the same inputs as item-level evaluators (input, output, expected_output, metadata)
2589                plus the list of evaluations from item-level evaluators. Useful for weighted averages,
2590                pass/fail decisions based on multiple criteria, or custom scoring logic combining multiple metrics.
2591            run_evaluators: List of functions to evaluate the entire experiment run.
2592                Each run evaluator receives all item_results and can compute aggregate metrics.
2593                Useful for calculating averages, distributions, or cross-item comparisons.
2594            max_concurrency: Maximum number of concurrent task executions (default: 50).
2595                Controls the number of items processed simultaneously. Adjust based on
2596                API rate limits and system resources.
2597            metadata: Optional metadata dictionary to attach to all experiment traces.
2598                This metadata will be included in every trace created during the experiment.
2599                If `data` are Langfuse dataset items, the metadata will be attached to the dataset run, too.
2600
2601        Returns:
2602            ExperimentResult containing:
2603            - run_name: The experiment run name. This is equal to the dataset run name if experiment was on Langfuse dataset.
2604            - item_results: List of results for each processed item with outputs and evaluations
2605            - run_evaluations: List of aggregate evaluation results for the entire run
2606            - dataset_run_id: ID of the dataset run (if using Langfuse datasets)
2607            - dataset_run_url: Direct URL to view results in Langfuse UI (if applicable)
2608
2609        Raises:
2610            ValueError: If required parameters are missing or invalid
2611            Exception: If experiment setup fails (individual item failures are handled gracefully)
2612
2613        Examples:
2614            Basic experiment with local data:
2615            ```python
2616            def summarize_text(*, item, **kwargs):
2617                return f"Summary: {item['input'][:50]}..."
2618
2619            def length_evaluator(*, input, output, expected_output=None, **kwargs):
2620                return {
2621                    "name": "output_length",
2622                    "value": len(output),
2623                    "comment": f"Output contains {len(output)} characters"
2624                }
2625
2626            result = langfuse.run_experiment(
2627                name="Text Summarization Test",
2628                description="Evaluate summarization quality and length",
2629                data=[
2630                    {"input": "Long article text...", "expected_output": "Expected summary"},
2631                    {"input": "Another article...", "expected_output": "Another summary"}
2632                ],
2633                task=summarize_text,
2634                evaluators=[length_evaluator]
2635            )
2636
2637            print(f"Processed {len(result.item_results)} items")
2638            for item_result in result.item_results:
2639                print(f"Input: {item_result.item['input']}")
2640                print(f"Output: {item_result.output}")
2641                print(f"Evaluations: {item_result.evaluations}")
2642            ```
2643
2644            Advanced experiment with async task and multiple evaluators:
2645            ```python
2646            async def llm_task(*, item, **kwargs):
2647                # Simulate async LLM call
2648                response = await openai_client.chat.completions.create(
2649                    model="gpt-4",
2650                    messages=[{"role": "user", "content": item["input"]}]
2651                )
2652                return response.choices[0].message.content
2653
2654            def accuracy_evaluator(*, input, output, expected_output=None, **kwargs):
2655                if expected_output and expected_output.lower() in output.lower():
2656                    return {"name": "accuracy", "value": 1.0, "comment": "Correct answer"}
2657                return {"name": "accuracy", "value": 0.0, "comment": "Incorrect answer"}
2658
2659            def toxicity_evaluator(*, input, output, expected_output=None, **kwargs):
2660                # Simulate toxicity check
2661                toxicity_score = check_toxicity(output)  # Your toxicity checker
2662                return {
2663                    "name": "toxicity",
2664                    "value": toxicity_score,
2665                    "comment": f"Toxicity level: {'high' if toxicity_score > 0.7 else 'low'}"
2666                }
2667
2668            def average_accuracy(*, item_results, **kwargs):
2669                accuracies = [
2670                    eval.value for result in item_results
2671                    for eval in result.evaluations
2672                    if eval.name == "accuracy"
2673                ]
2674                return {
2675                    "name": "average_accuracy",
2676                    "value": sum(accuracies) / len(accuracies) if accuracies else 0,
2677                    "comment": f"Average accuracy across {len(accuracies)} items"
2678                }
2679
2680            result = langfuse.run_experiment(
2681                name="LLM Safety and Accuracy Test",
2682                description="Evaluate model accuracy and safety across diverse prompts",
2683                data=test_dataset,  # Your dataset items
2684                task=llm_task,
2685                evaluators=[accuracy_evaluator, toxicity_evaluator],
2686                run_evaluators=[average_accuracy],
2687                max_concurrency=5,  # Limit concurrent API calls
2688                metadata={"model": "gpt-4", "temperature": 0.7}
2689            )
2690            ```
2691
2692            Using with Langfuse datasets:
2693            ```python
2694            # Get dataset from Langfuse
2695            dataset = langfuse.get_dataset("my-eval-dataset")
2696
2697            result = dataset.run_experiment(
2698                name="Production Model Evaluation",
2699                description="Monthly evaluation of production model performance",
2700                task=my_production_task,
2701                evaluators=[accuracy_evaluator, latency_evaluator]
2702            )
2703
2704            # Results automatically linked to dataset in Langfuse UI
2705            print(f"View results: {result['dataset_run_url']}")
2706            ```
2707
2708        Note:
2709            - Task and evaluator functions can be either synchronous or asynchronous
2710            - Individual item failures are logged but don't stop the experiment
2711            - All executions are automatically traced and visible in Langfuse UI
2712            - When using Langfuse datasets, results are automatically linked for easy comparison
2713            - This method works in both sync and async contexts (Jupyter notebooks, web apps, etc.)
2714            - Async execution is handled automatically with smart event loop detection
2715        """
2716        return cast(
2717            ExperimentResult,
2718            run_async_safely(
2719                self._run_experiment_async(
2720                    name=name,
2721                    run_name=self._create_experiment_run_name(
2722                        name=name, run_name=run_name
2723                    ),
2724                    description=description,
2725                    data=data,
2726                    task=task,
2727                    evaluators=evaluators or [],
2728                    composite_evaluator=composite_evaluator,
2729                    run_evaluators=run_evaluators or [],
2730                    max_concurrency=max_concurrency,
2731                    metadata=metadata,
2732                ),
2733            ),
2734        )
2735
2736    async def _run_experiment_async(
2737        self,
2738        *,
2739        name: str,
2740        run_name: str,
2741        description: Optional[str],
2742        data: ExperimentData,
2743        task: TaskFunction,
2744        evaluators: List[EvaluatorFunction],
2745        composite_evaluator: Optional[CompositeEvaluatorFunction],
2746        run_evaluators: List[RunEvaluatorFunction],
2747        max_concurrency: int,
2748        metadata: Optional[Dict[str, Any]] = None,
2749    ) -> ExperimentResult:
2750        langfuse_logger.debug(
2751            f"Starting experiment '{name}' run '{run_name}' with {len(data)} items"
2752        )
2753
2754        # Set up concurrency control
2755        semaphore = asyncio.Semaphore(max_concurrency)
2756
2757        # Process all items
2758        async def process_item(item: ExperimentItem) -> ExperimentItemResult:
2759            async with semaphore:
2760                return await self._process_experiment_item(
2761                    item,
2762                    task,
2763                    evaluators,
2764                    composite_evaluator,
2765                    name,
2766                    run_name,
2767                    description,
2768                    metadata,
2769                )
2770
2771        # Run all items concurrently
2772        tasks = [process_item(item) for item in data]
2773        item_results = await asyncio.gather(*tasks, return_exceptions=True)
2774
2775        # Filter out any exceptions and log errors
2776        valid_results: List[ExperimentItemResult] = []
2777        for i, result in enumerate(item_results):
2778            if isinstance(result, Exception):
2779                langfuse_logger.error(f"Item {i} failed: {result}")
2780            elif isinstance(result, ExperimentItemResult):
2781                valid_results.append(result)  # type: ignore
2782
2783        # Run experiment-level evaluators
2784        run_evaluations: List[Evaluation] = []
2785        for run_evaluator in run_evaluators:
2786            try:
2787                evaluations = await _run_evaluator(
2788                    run_evaluator, item_results=valid_results
2789                )
2790                run_evaluations.extend(evaluations)
2791            except Exception as e:
2792                langfuse_logger.error(f"Run evaluator failed: {e}")
2793
2794        # Generate dataset run URL if applicable
2795        dataset_run_id = valid_results[0].dataset_run_id if valid_results else None
2796        dataset_run_url = None
2797        if dataset_run_id and data:
2798            try:
2799                # Check if the first item has dataset_id (for DatasetItem objects)
2800                first_item = data[0]
2801                dataset_id = None
2802
2803                if hasattr(first_item, "dataset_id"):
2804                    dataset_id = getattr(first_item, "dataset_id", None)
2805
2806                if dataset_id:
2807                    project_id = self._get_project_id()
2808
2809                    if project_id:
2810                        dataset_run_url = f"{self._base_url}/project/{project_id}/datasets/{dataset_id}/runs/{dataset_run_id}"
2811
2812            except Exception:
2813                pass  # URL generation is optional
2814
2815        # Store run-level evaluations as scores
2816        for evaluation in run_evaluations:
2817            try:
2818                if dataset_run_id:
2819                    self.create_score(
2820                        dataset_run_id=dataset_run_id,
2821                        name=evaluation.name or "<unknown>",
2822                        value=evaluation.value,  # type: ignore
2823                        comment=evaluation.comment,
2824                        metadata=evaluation.metadata,
2825                        data_type=evaluation.data_type,  # type: ignore
2826                        config_id=evaluation.config_id,
2827                    )
2828
2829            except Exception as e:
2830                langfuse_logger.error(f"Failed to store run evaluation: {e}")
2831
2832        # Flush scores and traces
2833        self.flush()
2834
2835        return ExperimentResult(
2836            name=name,
2837            run_name=run_name,
2838            description=description,
2839            item_results=valid_results,
2840            run_evaluations=run_evaluations,
2841            dataset_run_id=dataset_run_id,
2842            dataset_run_url=dataset_run_url,
2843        )
2844
2845    async def _process_experiment_item(
2846        self,
2847        item: ExperimentItem,
2848        task: Callable,
2849        evaluators: List[Callable],
2850        composite_evaluator: Optional[CompositeEvaluatorFunction],
2851        experiment_name: str,
2852        experiment_run_name: str,
2853        experiment_description: Optional[str],
2854        experiment_metadata: Optional[Dict[str, Any]] = None,
2855    ) -> ExperimentItemResult:
2856        span_name = "experiment-item-run"
2857
2858        with self.start_as_current_span(name=span_name) as span:
2859            try:
2860                input_data = (
2861                    item.get("input")
2862                    if isinstance(item, dict)
2863                    else getattr(item, "input", None)
2864                )
2865
2866                if input_data is None:
2867                    raise ValueError("Experiment Item is missing input. Skipping item.")
2868
2869                expected_output = (
2870                    item.get("expected_output")
2871                    if isinstance(item, dict)
2872                    else getattr(item, "expected_output", None)
2873                )
2874
2875                item_metadata = (
2876                    item.get("metadata")
2877                    if isinstance(item, dict)
2878                    else getattr(item, "metadata", None)
2879                )
2880
2881                final_observation_metadata = {
2882                    "experiment_name": experiment_name,
2883                    "experiment_run_name": experiment_run_name,
2884                    **(experiment_metadata or {}),
2885                }
2886
2887                trace_id = span.trace_id
2888                dataset_id = None
2889                dataset_item_id = None
2890                dataset_run_id = None
2891
2892                # Link to dataset run if this is a dataset item
2893                if hasattr(item, "id") and hasattr(item, "dataset_id"):
2894                    try:
2895                        # Use sync API to avoid event loop issues when run_async_safely
2896                        # creates multiple event loops across different threads
2897                        dataset_run_item = await asyncio.to_thread(
2898                            self.api.dataset_run_items.create,
2899                            request=CreateDatasetRunItemRequest(
2900                                runName=experiment_run_name,
2901                                runDescription=experiment_description,
2902                                metadata=experiment_metadata,
2903                                datasetItemId=item.id,  # type: ignore
2904                                traceId=trace_id,
2905                                observationId=span.id,
2906                            ),
2907                        )
2908
2909                        dataset_run_id = dataset_run_item.dataset_run_id
2910
2911                    except Exception as e:
2912                        langfuse_logger.error(f"Failed to create dataset run item: {e}")
2913
2914                if (
2915                    not isinstance(item, dict)
2916                    and hasattr(item, "dataset_id")
2917                    and hasattr(item, "id")
2918                ):
2919                    dataset_id = item.dataset_id
2920                    dataset_item_id = item.id
2921
2922                    final_observation_metadata.update(
2923                        {"dataset_id": dataset_id, "dataset_item_id": dataset_item_id}
2924                    )
2925
2926                if isinstance(item_metadata, dict):
2927                    final_observation_metadata.update(item_metadata)
2928
2929                experiment_id = dataset_run_id or self._create_observation_id()
2930                experiment_item_id = (
2931                    dataset_item_id or get_sha256_hash_hex(_serialize(input_data))[:16]
2932                )
2933                span._otel_span.set_attributes(
2934                    {
2935                        k: v
2936                        for k, v in {
2937                            LangfuseOtelSpanAttributes.ENVIRONMENT: LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT,
2938                            LangfuseOtelSpanAttributes.EXPERIMENT_DESCRIPTION: experiment_description,
2939                            LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_EXPECTED_OUTPUT: _serialize(
2940                                expected_output
2941                            ),
2942                        }.items()
2943                        if v is not None
2944                    }
2945                )
2946
2947                propagated_experiment_attributes = PropagatedExperimentAttributes(
2948                    experiment_id=experiment_id,
2949                    experiment_name=experiment_run_name,
2950                    experiment_metadata=_serialize(experiment_metadata),
2951                    experiment_dataset_id=dataset_id,
2952                    experiment_item_id=experiment_item_id,
2953                    experiment_item_metadata=_serialize(item_metadata),
2954                    experiment_item_root_observation_id=span.id,
2955                )
2956
2957                with _propagate_attributes(experiment=propagated_experiment_attributes):
2958                    output = await _run_task(task, item)
2959
2960                span.update(
2961                    input=input_data,
2962                    output=output,
2963                    metadata=final_observation_metadata,
2964                )
2965
2966            except Exception as e:
2967                span.update(
2968                    output=f"Error: {str(e)}", level="ERROR", status_message=str(e)
2969                )
2970                raise e
2971
2972            # Run evaluators
2973            evaluations = []
2974
2975            for evaluator in evaluators:
2976                try:
2977                    eval_metadata: Optional[Dict[str, Any]] = None
2978
2979                    if isinstance(item, dict):
2980                        eval_metadata = item.get("metadata")
2981                    elif hasattr(item, "metadata"):
2982                        eval_metadata = item.metadata
2983
2984                    with _propagate_attributes(
2985                        experiment=propagated_experiment_attributes
2986                    ):
2987                        eval_results = await _run_evaluator(
2988                            evaluator,
2989                            input=input_data,
2990                            output=output,
2991                            expected_output=expected_output,
2992                            metadata=eval_metadata,
2993                        )
2994                        evaluations.extend(eval_results)
2995
2996                        # Store evaluations as scores
2997                        for evaluation in eval_results:
2998                            self.create_score(
2999                                trace_id=trace_id,
3000                                observation_id=span.id,
3001                                name=evaluation.name,
3002                                value=evaluation.value,  # type: ignore
3003                                comment=evaluation.comment,
3004                                metadata=evaluation.metadata,
3005                                config_id=evaluation.config_id,
3006                                data_type=evaluation.data_type,  # type: ignore
3007                            )
3008
3009                except Exception as e:
3010                    langfuse_logger.error(f"Evaluator failed: {e}")
3011
3012            # Run composite evaluator if provided and we have evaluations
3013            if composite_evaluator and evaluations:
3014                try:
3015                    composite_eval_metadata: Optional[Dict[str, Any]] = None
3016                    if isinstance(item, dict):
3017                        composite_eval_metadata = item.get("metadata")
3018                    elif hasattr(item, "metadata"):
3019                        composite_eval_metadata = item.metadata
3020
3021                    with _propagate_attributes(
3022                        experiment=propagated_experiment_attributes
3023                    ):
3024                        result = composite_evaluator(
3025                            input=input_data,
3026                            output=output,
3027                            expected_output=expected_output,
3028                            metadata=composite_eval_metadata,
3029                            evaluations=evaluations,
3030                        )
3031
3032                        # Handle async composite evaluators
3033                        if asyncio.iscoroutine(result):
3034                            result = await result
3035
3036                        # Normalize to list
3037                        composite_evals: List[Evaluation] = []
3038                        if isinstance(result, (dict, Evaluation)):
3039                            composite_evals = [result]  # type: ignore
3040                        elif isinstance(result, list):
3041                            composite_evals = result  # type: ignore
3042
3043                        # Store composite evaluations as scores and add to evaluations list
3044                        for composite_evaluation in composite_evals:
3045                            self.create_score(
3046                                trace_id=trace_id,
3047                                observation_id=span.id,
3048                                name=composite_evaluation.name,
3049                                value=composite_evaluation.value,  # type: ignore
3050                                comment=composite_evaluation.comment,
3051                                metadata=composite_evaluation.metadata,
3052                                config_id=composite_evaluation.config_id,
3053                                data_type=composite_evaluation.data_type,  # type: ignore
3054                            )
3055                            evaluations.append(composite_evaluation)
3056
3057                except Exception as e:
3058                    langfuse_logger.error(f"Composite evaluator failed: {e}")
3059
3060            return ExperimentItemResult(
3061                item=item,
3062                output=output,
3063                evaluations=evaluations,
3064                trace_id=trace_id,
3065                dataset_run_id=dataset_run_id,
3066            )
3067
3068    def _create_experiment_run_name(
3069        self, *, name: Optional[str] = None, run_name: Optional[str] = None
3070    ) -> str:
3071        if run_name:
3072            return run_name
3073
3074        iso_timestamp = _get_timestamp().isoformat().replace("+00:00", "Z")
3075
3076        return f"{name} - {iso_timestamp}"
3077
3078    def run_batched_evaluation(
3079        self,
3080        *,
3081        scope: Literal["traces", "observations"],
3082        mapper: MapperFunction,
3083        filter: Optional[str] = None,
3084        fetch_batch_size: int = 50,
3085        max_items: Optional[int] = None,
3086        max_retries: int = 3,
3087        evaluators: List[EvaluatorFunction],
3088        composite_evaluator: Optional[CompositeEvaluatorFunction] = None,
3089        max_concurrency: int = 50,
3090        metadata: Optional[Dict[str, Any]] = None,
3091        resume_from: Optional[BatchEvaluationResumeToken] = None,
3092        verbose: bool = False,
3093    ) -> BatchEvaluationResult:
3094        """Fetch traces or observations and run evaluations on each item.
3095
3096        This method provides a powerful way to evaluate existing data in Langfuse at scale.
3097        It fetches items based on filters, transforms them using a mapper function, runs
3098        evaluators on each item, and creates scores that are linked back to the original
3099        entities. This is ideal for:
3100
3101        - Running evaluations on production traces after deployment
3102        - Backtesting new evaluation metrics on historical data
3103        - Batch scoring of observations for quality monitoring
3104        - Periodic evaluation runs on recent data
3105
3106        The method uses a streaming/pipeline approach to process items in batches, making
3107        it memory-efficient for large datasets. It includes comprehensive error handling,
3108        retry logic, and resume capability for long-running evaluations.
3109
3110        Args:
3111            scope: The type of items to evaluate. Must be one of:
3112                - "traces": Evaluate complete traces with all their observations
3113                - "observations": Evaluate individual observations (spans, generations, events)
3114            mapper: Function that transforms API response objects into evaluator inputs.
3115                Receives a trace/observation object and returns an EvaluatorInputs
3116                instance with input, output, expected_output, and metadata fields.
3117                Can be sync or async.
3118            evaluators: List of evaluation functions to run on each item. Each evaluator
3119                receives the mapped inputs and returns Evaluation object(s). Evaluator
3120                failures are logged but don't stop the batch evaluation.
3121            filter: Optional JSON filter string for querying items (same format as Langfuse API). Examples:
3122                - '{"tags": ["production"]}'
3123                - '{"user_id": "user123", "timestamp": {"operator": ">", "value": "2024-01-01"}}'
3124                Default: None (fetches all items).
3125            fetch_batch_size: Number of items to fetch per API call and hold in memory.
3126                Larger values may be faster but use more memory. Default: 50.
3127            max_items: Maximum total number of items to process. If None, processes all
3128                items matching the filter. Useful for testing or limiting evaluation runs.
3129                Default: None (process all).
3130            max_concurrency: Maximum number of items to evaluate concurrently. Controls
3131                parallelism and resource usage. Default: 50.
3132            composite_evaluator: Optional function that creates a composite score from
3133                item-level evaluations. Receives the original item and its evaluations,
3134                returns a single Evaluation. Useful for weighted averages or combined metrics.
3135                Default: None.
3136            metadata: Optional metadata dict to add to all created scores. Useful for
3137                tracking evaluation runs, versions, or other context. Default: None.
3138            max_retries: Maximum number of retry attempts for failed batch fetches.
3139                Uses exponential backoff (1s, 2s, 4s). Default: 3.
3140            verbose: If True, logs progress information to console. Useful for monitoring
3141                long-running evaluations. Default: False.
3142            resume_from: Optional resume token from a previous incomplete run. Allows
3143                continuing evaluation after interruption or failure. Default: None.
3144
3145
3146        Returns:
3147            BatchEvaluationResult containing:
3148                - total_items_fetched: Number of items fetched from API
3149                - total_items_processed: Number of items successfully evaluated
3150                - total_items_failed: Number of items that failed evaluation
3151                - total_scores_created: Scores created by item-level evaluators
3152                - total_composite_scores_created: Scores created by composite evaluator
3153                - total_evaluations_failed: Individual evaluator failures
3154                - evaluator_stats: Per-evaluator statistics (success rate, scores created)
3155                - resume_token: Token for resuming if incomplete (None if completed)
3156                - completed: True if all items processed
3157                - duration_seconds: Total execution time
3158                - failed_item_ids: IDs of items that failed
3159                - error_summary: Error types and counts
3160                - has_more_items: True if max_items reached but more exist
3161
3162        Raises:
3163            ValueError: If invalid scope is provided.
3164
3165        Examples:
3166            Basic trace evaluation:
3167            ```python
3168            from langfuse import Langfuse, EvaluatorInputs, Evaluation
3169
3170            client = Langfuse()
3171
3172            # Define mapper to extract fields from traces
3173            def trace_mapper(trace):
3174                return EvaluatorInputs(
3175                    input=trace.input,
3176                    output=trace.output,
3177                    expected_output=None,
3178                    metadata={"trace_id": trace.id}
3179                )
3180
3181            # Define evaluator
3182            def length_evaluator(*, input, output, expected_output, metadata):
3183                return Evaluation(
3184                    name="output_length",
3185                    value=len(output) if output else 0
3186                )
3187
3188            # Run batch evaluation
3189            result = client.run_batched_evaluation(
3190                scope="traces",
3191                mapper=trace_mapper,
3192                evaluators=[length_evaluator],
3193                filter='{"tags": ["production"]}',
3194                max_items=1000,
3195                verbose=True
3196            )
3197
3198            print(f"Processed {result.total_items_processed} traces")
3199            print(f"Created {result.total_scores_created} scores")
3200            ```
3201
3202            Evaluation with composite scorer:
3203            ```python
3204            def accuracy_evaluator(*, input, output, expected_output, metadata):
3205                # ... evaluation logic
3206                return Evaluation(name="accuracy", value=0.85)
3207
3208            def relevance_evaluator(*, input, output, expected_output, metadata):
3209                # ... evaluation logic
3210                return Evaluation(name="relevance", value=0.92)
3211
3212            def composite_evaluator(*, item, evaluations):
3213                # Weighted average of evaluations
3214                weights = {"accuracy": 0.6, "relevance": 0.4}
3215                total = sum(
3216                    e.value * weights.get(e.name, 0)
3217                    for e in evaluations
3218                    if isinstance(e.value, (int, float))
3219                )
3220                return Evaluation(
3221                    name="composite_score",
3222                    value=total,
3223                    comment=f"Weighted average of {len(evaluations)} metrics"
3224                )
3225
3226            result = client.run_batched_evaluation(
3227                scope="traces",
3228                mapper=trace_mapper,
3229                evaluators=[accuracy_evaluator, relevance_evaluator],
3230                composite_evaluator=composite_evaluator,
3231                filter='{"user_id": "important_user"}',
3232                verbose=True
3233            )
3234            ```
3235
3236            Handling incomplete runs with resume:
3237            ```python
3238            # Initial run that may fail or timeout
3239            result = client.run_batched_evaluation(
3240                scope="observations",
3241                mapper=obs_mapper,
3242                evaluators=[my_evaluator],
3243                max_items=10000,
3244                verbose=True
3245            )
3246
3247            # Check if incomplete
3248            if not result.completed and result.resume_token:
3249                print(f"Processed {result.resume_token.items_processed} items before interruption")
3250
3251                # Resume from where it left off
3252                result = client.run_batched_evaluation(
3253                    scope="observations",
3254                    mapper=obs_mapper,
3255                    evaluators=[my_evaluator],
3256                    resume_from=result.resume_token,
3257                    verbose=True
3258                )
3259
3260            print(f"Total items processed: {result.total_items_processed}")
3261            ```
3262
3263            Monitoring evaluator performance:
3264            ```python
3265            result = client.run_batched_evaluation(...)
3266
3267            for stats in result.evaluator_stats:
3268                success_rate = stats.successful_runs / stats.total_runs
3269                print(f"{stats.name}:")
3270                print(f"  Success rate: {success_rate:.1%}")
3271                print(f"  Scores created: {stats.total_scores_created}")
3272
3273                if stats.failed_runs > 0:
3274                    print(f"  ⚠️  Failed {stats.failed_runs} times")
3275            ```
3276
3277        Note:
3278            - Evaluator failures are logged but don't stop the batch evaluation
3279            - Individual item failures are tracked but don't stop processing
3280            - Fetch failures are retried with exponential backoff
3281            - All scores are automatically flushed to Langfuse at the end
3282            - The resume mechanism uses timestamp-based filtering to avoid duplicates
3283        """
3284        runner = BatchEvaluationRunner(self)
3285
3286        return cast(
3287            BatchEvaluationResult,
3288            run_async_safely(
3289                runner.run_async(
3290                    scope=scope,
3291                    mapper=mapper,
3292                    evaluators=evaluators,
3293                    filter=filter,
3294                    fetch_batch_size=fetch_batch_size,
3295                    max_items=max_items,
3296                    max_concurrency=max_concurrency,
3297                    composite_evaluator=composite_evaluator,
3298                    metadata=metadata,
3299                    max_retries=max_retries,
3300                    verbose=verbose,
3301                    resume_from=resume_from,
3302                )
3303            ),
3304        )
3305
3306    def auth_check(self) -> bool:
3307        """Check if the provided credentials (public and secret key) are valid.
3308
3309        Raises:
3310            Exception: If no projects were found for the provided credentials.
3311
3312        Note:
3313            This method is blocking. It is discouraged to use it in production code.
3314        """
3315        try:
3316            projects = self.api.projects.get()
3317            langfuse_logger.debug(
3318                f"Auth check successful, found {len(projects.data)} projects"
3319            )
3320            if len(projects.data) == 0:
3321                raise Exception(
3322                    "Auth check failed, no project found for the keys provided."
3323                )
3324            return True
3325
3326        except AttributeError as e:
3327            langfuse_logger.warning(
3328                f"Auth check failed: Client not properly initialized. Error: {e}"
3329            )
3330            return False
3331
3332        except Error as e:
3333            handle_fern_exception(e)
3334            raise e
3335
3336    def create_dataset(
3337        self,
3338        *,
3339        name: str,
3340        description: Optional[str] = None,
3341        metadata: Optional[Any] = None,
3342        input_schema: Optional[Any] = None,
3343        expected_output_schema: Optional[Any] = None,
3344    ) -> Dataset:
3345        """Create a dataset with the given name on Langfuse.
3346
3347        Args:
3348            name: Name of the dataset to create.
3349            description: Description of the dataset. Defaults to None.
3350            metadata: Additional metadata. Defaults to None.
3351            input_schema: JSON Schema for validating dataset item inputs. When set, all new items will be validated against this schema.
3352            expected_output_schema: JSON Schema for validating dataset item expected outputs. When set, all new items will be validated against this schema.
3353
3354        Returns:
3355            Dataset: The created dataset as returned by the Langfuse API.
3356        """
3357        try:
3358            body = CreateDatasetRequest(
3359                name=name,
3360                description=description,
3361                metadata=metadata,
3362                inputSchema=input_schema,
3363                expectedOutputSchema=expected_output_schema,
3364            )
3365            langfuse_logger.debug(f"Creating datasets {body}")
3366
3367            return self.api.datasets.create(request=body)
3368
3369        except Error as e:
3370            handle_fern_exception(e)
3371            raise e
3372
3373    def create_dataset_item(
3374        self,
3375        *,
3376        dataset_name: str,
3377        input: Optional[Any] = None,
3378        expected_output: Optional[Any] = None,
3379        metadata: Optional[Any] = None,
3380        source_trace_id: Optional[str] = None,
3381        source_observation_id: Optional[str] = None,
3382        status: Optional[DatasetStatus] = None,
3383        id: Optional[str] = None,
3384    ) -> DatasetItem:
3385        """Create a dataset item.
3386
3387        Upserts if an item with id already exists.
3388
3389        Args:
3390            dataset_name: Name of the dataset in which the dataset item should be created.
3391            input: Input data. Defaults to None. Can contain any dict, list or scalar.
3392            expected_output: Expected output data. Defaults to None. Can contain any dict, list or scalar.
3393            metadata: Additional metadata. Defaults to None. Can contain any dict, list or scalar.
3394            source_trace_id: Id of the source trace. Defaults to None.
3395            source_observation_id: Id of the source observation. Defaults to None.
3396            status: Status of the dataset item. Defaults to ACTIVE for newly created items.
3397            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.
3398
3399        Returns:
3400            DatasetItem: The created dataset item as returned by the Langfuse API.
3401
3402        Example:
3403            ```python
3404            from langfuse import Langfuse
3405
3406            langfuse = Langfuse()
3407
3408            # Uploading items to the Langfuse dataset named "capital_cities"
3409            langfuse.create_dataset_item(
3410                dataset_name="capital_cities",
3411                input={"input": {"country": "Italy"}},
3412                expected_output={"expected_output": "Rome"},
3413                metadata={"foo": "bar"}
3414            )
3415            ```
3416        """
3417        try:
3418            body = CreateDatasetItemRequest(
3419                datasetName=dataset_name,
3420                input=input,
3421                expectedOutput=expected_output,
3422                metadata=metadata,
3423                sourceTraceId=source_trace_id,
3424                sourceObservationId=source_observation_id,
3425                status=status,
3426                id=id,
3427            )
3428            langfuse_logger.debug(f"Creating dataset item {body}")
3429            return self.api.dataset_items.create(request=body)
3430        except Error as e:
3431            handle_fern_exception(e)
3432            raise e
3433
3434    def resolve_media_references(
3435        self,
3436        *,
3437        obj: Any,
3438        resolve_with: Literal["base64_data_uri"],
3439        max_depth: int = 10,
3440        content_fetch_timeout_seconds: int = 5,
3441    ) -> Any:
3442        """Replace media reference strings in an object with base64 data URIs.
3443
3444        This method recursively traverses an object (up to max_depth) looking for media reference strings
3445        in the format "@@@langfuseMedia:...@@@". When found, it (synchronously) fetches the actual media content using
3446        the provided Langfuse client and replaces the reference string with a base64 data URI.
3447
3448        If fetching media content fails for a reference string, a warning is logged and the reference
3449        string is left unchanged.
3450
3451        Args:
3452            obj: The object to process. Can be a primitive value, array, or nested object.
3453                If the object has a __dict__ attribute, a dict will be returned instead of the original object type.
3454            resolve_with: The representation of the media content to replace the media reference string with.
3455                Currently only "base64_data_uri" is supported.
3456            max_depth: int: The maximum depth to traverse the object. Default is 10.
3457            content_fetch_timeout_seconds: int: The timeout in seconds for fetching media content. Default is 5.
3458
3459        Returns:
3460            A deep copy of the input object with all media references replaced with base64 data URIs where possible.
3461            If the input object has a __dict__ attribute, a dict will be returned instead of the original object type.
3462
3463        Example:
3464            obj = {
3465                "image": "@@@langfuseMedia:type=image/jpeg|id=123|source=bytes@@@",
3466                "nested": {
3467                    "pdf": "@@@langfuseMedia:type=application/pdf|id=456|source=bytes@@@"
3468                }
3469            }
3470
3471            result = await LangfuseMedia.resolve_media_references(obj, langfuse_client)
3472
3473            # Result:
3474            # {
3475            #     "image": "data:image/jpeg;base64,/9j/4AAQSkZJRg...",
3476            #     "nested": {
3477            #         "pdf": "data:application/pdf;base64,JVBERi0xLjcK..."
3478            #     }
3479            # }
3480        """
3481        return LangfuseMedia.resolve_media_references(
3482            langfuse_client=self,
3483            obj=obj,
3484            resolve_with=resolve_with,
3485            max_depth=max_depth,
3486            content_fetch_timeout_seconds=content_fetch_timeout_seconds,
3487        )
3488
3489    @overload
3490    def get_prompt(
3491        self,
3492        name: str,
3493        *,
3494        version: Optional[int] = None,
3495        label: Optional[str] = None,
3496        type: Literal["chat"],
3497        cache_ttl_seconds: Optional[int] = None,
3498        fallback: Optional[List[ChatMessageDict]] = None,
3499        max_retries: Optional[int] = None,
3500        fetch_timeout_seconds: Optional[int] = None,
3501    ) -> ChatPromptClient: ...
3502
3503    @overload
3504    def get_prompt(
3505        self,
3506        name: str,
3507        *,
3508        version: Optional[int] = None,
3509        label: Optional[str] = None,
3510        type: Literal["text"] = "text",
3511        cache_ttl_seconds: Optional[int] = None,
3512        fallback: Optional[str] = None,
3513        max_retries: Optional[int] = None,
3514        fetch_timeout_seconds: Optional[int] = None,
3515    ) -> TextPromptClient: ...
3516
3517    def get_prompt(
3518        self,
3519        name: str,
3520        *,
3521        version: Optional[int] = None,
3522        label: Optional[str] = None,
3523        type: Literal["chat", "text"] = "text",
3524        cache_ttl_seconds: Optional[int] = None,
3525        fallback: Union[Optional[List[ChatMessageDict]], Optional[str]] = None,
3526        max_retries: Optional[int] = None,
3527        fetch_timeout_seconds: Optional[int] = None,
3528    ) -> PromptClient:
3529        """Get a prompt.
3530
3531        This method attempts to fetch the requested prompt from the local cache. If the prompt is not found
3532        in the cache or if the cached prompt has expired, it will try to fetch the prompt from the server again
3533        and update the cache. If fetching the new prompt fails, and there is an expired prompt in the cache, it will
3534        return the expired prompt as a fallback.
3535
3536        Args:
3537            name (str): The name of the prompt to retrieve.
3538
3539        Keyword Args:
3540            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.
3541            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.
3542            cache_ttl_seconds: Optional[int]: Time-to-live in seconds for caching the prompt. Must be specified as a
3543            keyword argument. If not set, defaults to 60 seconds. Disables caching if set to 0.
3544            type: Literal["chat", "text"]: The type of the prompt to retrieve. Defaults to "text".
3545            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.
3546            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.
3547            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.
3548
3549        Returns:
3550            The prompt object retrieved from the cache or directly fetched if not cached or expired of type
3551            - TextPromptClient, if type argument is 'text'.
3552            - ChatPromptClient, if type argument is 'chat'.
3553
3554        Raises:
3555            Exception: Propagates any exceptions raised during the fetching of a new prompt, unless there is an
3556            expired prompt in the cache, in which case it logs a warning and returns the expired prompt.
3557        """
3558        if self._resources is None:
3559            raise Error(
3560                "SDK is not correctly initialized. Check the init logs for more details."
3561            )
3562        if version is not None and label is not None:
3563            raise ValueError("Cannot specify both version and label at the same time.")
3564
3565        if not name:
3566            raise ValueError("Prompt name cannot be empty.")
3567
3568        cache_key = PromptCache.generate_cache_key(name, version=version, label=label)
3569        bounded_max_retries = self._get_bounded_max_retries(
3570            max_retries, default_max_retries=2, max_retries_upper_bound=4
3571        )
3572
3573        langfuse_logger.debug(f"Getting prompt '{cache_key}'")
3574        cached_prompt = self._resources.prompt_cache.get(cache_key)
3575
3576        if cached_prompt is None or cache_ttl_seconds == 0:
3577            langfuse_logger.debug(
3578                f"Prompt '{cache_key}' not found in cache or caching disabled."
3579            )
3580            try:
3581                return self._fetch_prompt_and_update_cache(
3582                    name,
3583                    version=version,
3584                    label=label,
3585                    ttl_seconds=cache_ttl_seconds,
3586                    max_retries=bounded_max_retries,
3587                    fetch_timeout_seconds=fetch_timeout_seconds,
3588                )
3589            except Exception as e:
3590                if fallback:
3591                    langfuse_logger.warning(
3592                        f"Returning fallback prompt for '{cache_key}' due to fetch error: {e}"
3593                    )
3594
3595                    fallback_client_args: Dict[str, Any] = {
3596                        "name": name,
3597                        "prompt": fallback,
3598                        "type": type,
3599                        "version": version or 0,
3600                        "config": {},
3601                        "labels": [label] if label else [],
3602                        "tags": [],
3603                    }
3604
3605                    if type == "text":
3606                        return TextPromptClient(
3607                            prompt=Prompt_Text(**fallback_client_args),
3608                            is_fallback=True,
3609                        )
3610
3611                    if type == "chat":
3612                        return ChatPromptClient(
3613                            prompt=Prompt_Chat(**fallback_client_args),
3614                            is_fallback=True,
3615                        )
3616
3617                raise e
3618
3619        if cached_prompt.is_expired():
3620            langfuse_logger.debug(f"Stale prompt '{cache_key}' found in cache.")
3621            try:
3622                # refresh prompt in background thread, refresh_prompt deduplicates tasks
3623                langfuse_logger.debug(f"Refreshing prompt '{cache_key}' in background.")
3624
3625                def refresh_task() -> None:
3626                    self._fetch_prompt_and_update_cache(
3627                        name,
3628                        version=version,
3629                        label=label,
3630                        ttl_seconds=cache_ttl_seconds,
3631                        max_retries=bounded_max_retries,
3632                        fetch_timeout_seconds=fetch_timeout_seconds,
3633                    )
3634
3635                self._resources.prompt_cache.add_refresh_prompt_task(
3636                    cache_key,
3637                    refresh_task,
3638                )
3639                langfuse_logger.debug(
3640                    f"Returning stale prompt '{cache_key}' from cache."
3641                )
3642                # return stale prompt
3643                return cached_prompt.value
3644
3645            except Exception as e:
3646                langfuse_logger.warning(
3647                    f"Error when refreshing cached prompt '{cache_key}', returning cached version. Error: {e}"
3648                )
3649                # creation of refresh prompt task failed, return stale prompt
3650                return cached_prompt.value
3651
3652        return cached_prompt.value
3653
3654    def _fetch_prompt_and_update_cache(
3655        self,
3656        name: str,
3657        *,
3658        version: Optional[int] = None,
3659        label: Optional[str] = None,
3660        ttl_seconds: Optional[int] = None,
3661        max_retries: int,
3662        fetch_timeout_seconds: Optional[int],
3663    ) -> PromptClient:
3664        cache_key = PromptCache.generate_cache_key(name, version=version, label=label)
3665        langfuse_logger.debug(f"Fetching prompt '{cache_key}' from server...")
3666
3667        try:
3668
3669            @backoff.on_exception(
3670                backoff.constant, Exception, max_tries=max_retries + 1, logger=None
3671            )
3672            def fetch_prompts() -> Any:
3673                return self.api.prompts.get(
3674                    self._url_encode(name),
3675                    version=version,
3676                    label=label,
3677                    request_options={
3678                        "timeout_in_seconds": fetch_timeout_seconds,
3679                    }
3680                    if fetch_timeout_seconds is not None
3681                    else None,
3682                )
3683
3684            prompt_response = fetch_prompts()
3685
3686            prompt: PromptClient
3687            if prompt_response.type == "chat":
3688                prompt = ChatPromptClient(prompt_response)
3689            else:
3690                prompt = TextPromptClient(prompt_response)
3691
3692            if self._resources is not None:
3693                self._resources.prompt_cache.set(cache_key, prompt, ttl_seconds)
3694
3695            return prompt
3696
3697        except NotFoundError as not_found_error:
3698            langfuse_logger.warning(
3699                f"Prompt '{cache_key}' not found during refresh, evicting from cache."
3700            )
3701            if self._resources is not None:
3702                self._resources.prompt_cache.delete(cache_key)
3703            raise not_found_error
3704
3705        except Exception as e:
3706            langfuse_logger.error(
3707                f"Error while fetching prompt '{cache_key}': {str(e)}"
3708            )
3709            raise e
3710
3711    def _get_bounded_max_retries(
3712        self,
3713        max_retries: Optional[int],
3714        *,
3715        default_max_retries: int = 2,
3716        max_retries_upper_bound: int = 4,
3717    ) -> int:
3718        if max_retries is None:
3719            return default_max_retries
3720
3721        bounded_max_retries = min(
3722            max(max_retries, 0),
3723            max_retries_upper_bound,
3724        )
3725
3726        return bounded_max_retries
3727
3728    @overload
3729    def create_prompt(
3730        self,
3731        *,
3732        name: str,
3733        prompt: List[Union[ChatMessageDict, ChatMessageWithPlaceholdersDict]],
3734        labels: List[str] = [],
3735        tags: Optional[List[str]] = None,
3736        type: Optional[Literal["chat"]],
3737        config: Optional[Any] = None,
3738        commit_message: Optional[str] = None,
3739    ) -> ChatPromptClient: ...
3740
3741    @overload
3742    def create_prompt(
3743        self,
3744        *,
3745        name: str,
3746        prompt: str,
3747        labels: List[str] = [],
3748        tags: Optional[List[str]] = None,
3749        type: Optional[Literal["text"]] = "text",
3750        config: Optional[Any] = None,
3751        commit_message: Optional[str] = None,
3752    ) -> TextPromptClient: ...
3753
3754    def create_prompt(
3755        self,
3756        *,
3757        name: str,
3758        prompt: Union[
3759            str, List[Union[ChatMessageDict, ChatMessageWithPlaceholdersDict]]
3760        ],
3761        labels: List[str] = [],
3762        tags: Optional[List[str]] = None,
3763        type: Optional[Literal["chat", "text"]] = "text",
3764        config: Optional[Any] = None,
3765        commit_message: Optional[str] = None,
3766    ) -> PromptClient:
3767        """Create a new prompt in Langfuse.
3768
3769        Keyword Args:
3770            name : The name of the prompt to be created.
3771            prompt : The content of the prompt to be created.
3772            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.
3773            labels: The labels of the prompt. Defaults to None. To create a default-served prompt, add the 'production' label.
3774            tags: The tags of the prompt. Defaults to None. Will be applied to all versions of the prompt.
3775            config: Additional structured data to be saved with the prompt. Defaults to None.
3776            type: The type of the prompt to be created. "chat" vs. "text". Defaults to "text".
3777            commit_message: Optional string describing the change.
3778
3779        Returns:
3780            TextPromptClient: The prompt if type argument is 'text'.
3781            ChatPromptClient: The prompt if type argument is 'chat'.
3782        """
3783        try:
3784            langfuse_logger.debug(f"Creating prompt {name=}, {labels=}")
3785
3786            if type == "chat":
3787                if not isinstance(prompt, list):
3788                    raise ValueError(
3789                        "For 'chat' type, 'prompt' must be a list of chat messages with role and content attributes."
3790                    )
3791                request: Union[CreatePromptRequest_Chat, CreatePromptRequest_Text] = (
3792                    CreatePromptRequest_Chat(
3793                        name=name,
3794                        prompt=cast(Any, prompt),
3795                        labels=labels,
3796                        tags=tags,
3797                        config=config or {},
3798                        commitMessage=commit_message,
3799                        type="chat",
3800                    )
3801                )
3802                server_prompt = self.api.prompts.create(request=request)
3803
3804                if self._resources is not None:
3805                    self._resources.prompt_cache.invalidate(name)
3806
3807                return ChatPromptClient(prompt=cast(Prompt_Chat, server_prompt))
3808
3809            if not isinstance(prompt, str):
3810                raise ValueError("For 'text' type, 'prompt' must be a string.")
3811
3812            request = CreatePromptRequest_Text(
3813                name=name,
3814                prompt=prompt,
3815                labels=labels,
3816                tags=tags,
3817                config=config or {},
3818                commitMessage=commit_message,
3819                type="text",
3820            )
3821
3822            server_prompt = self.api.prompts.create(request=request)
3823
3824            if self._resources is not None:
3825                self._resources.prompt_cache.invalidate(name)
3826
3827            return TextPromptClient(prompt=cast(Prompt_Text, server_prompt))
3828
3829        except Error as e:
3830            handle_fern_exception(e)
3831            raise e
3832
3833    def update_prompt(
3834        self,
3835        *,
3836        name: str,
3837        version: int,
3838        new_labels: List[str] = [],
3839    ) -> Any:
3840        """Update an existing prompt version in Langfuse. The Langfuse SDK prompt cache is invalidated for all prompts witht he specified name.
3841
3842        Args:
3843            name (str): The name of the prompt to update.
3844            version (int): The version number of the prompt to update.
3845            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 [].
3846
3847        Returns:
3848            Prompt: The updated prompt from the Langfuse API.
3849
3850        """
3851        updated_prompt = self.api.prompt_version.update(
3852            name=self._url_encode(name),
3853            version=version,
3854            new_labels=new_labels,
3855        )
3856
3857        if self._resources is not None:
3858            self._resources.prompt_cache.invalidate(name)
3859
3860        return updated_prompt
3861
3862    def _url_encode(self, url: str, *, is_url_param: Optional[bool] = False) -> str:
3863        # httpx ≥ 0.28 does its own WHATWG-compliant quoting (eg. encodes bare
3864        # “%”, “?”, “#”, “|”, … in query/path parts).  Re-quoting here would
3865        # double-encode, so we skip when the value is about to be sent straight
3866        # to httpx (`is_url_param=True`) and the installed version is ≥ 0.28.
3867        if is_url_param and Version(httpx.__version__) >= Version("0.28.0"):
3868            return url
3869
3870        # urllib.parse.quote does not escape slashes "/" by default; we need to add safe="" to force escaping
3871        # we need add safe="" to force escaping of slashes
3872        # This is necessary for prompts in prompt folders
3873        return urllib.parse.quote(url, safe="")
3874
3875    def clear_prompt_cache(self) -> None:
3876        """Clear the entire prompt cache, removing all cached prompts.
3877
3878        This method is useful when you want to force a complete refresh of all
3879        cached prompts, for example after major updates or when you need to
3880        ensure the latest versions are fetched from the server.
3881        """
3882        if self._resources is not None:
3883            self._resources.prompt_cache.clear()