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