100class Langfuse:
 101    """Main client for Langfuse tracing and platform features.
 102
 103    This class provides an interface for creating and managing traces, spans,
 104    and generations in Langfuse as well as interacting with the Langfuse API.
 105
 106    The client features a thread-safe singleton pattern for each unique public API key,
 107    ensuring consistent trace context propagation across your application. It implements
 108    efficient batching of spans with configurable flush settings and includes background
 109    thread management for media uploads and score ingestion.
 110
 111    Configuration is flexible through either direct parameters or environment variables,
 112    with graceful fallbacks and runtime configuration updates.
 113
 114    Attributes:
 115        api: Synchronous API client for Langfuse backend communication
 116        async_api: Asynchronous API client for Langfuse backend communication
 117        langfuse_tracer: Internal LangfuseTracer instance managing OpenTelemetry components
 118
 119    Parameters:
 120        public_key (Optional[str]): Your Langfuse public API key. Can also be set via LANGFUSE_PUBLIC_KEY environment variable.
 121        secret_key (Optional[str]): Your Langfuse secret API key. Can also be set via LANGFUSE_SECRET_KEY environment variable.
 122        host (Optional[str]): The Langfuse API host URL. Defaults to "https://cloud.langfuse.com". Can also be set via LANGFUSE_HOST environment variable.
 123        timeout (Optional[int]): Timeout in seconds for API requests. Defaults to 5 seconds.
 124        httpx_client (Optional[httpx.Client]): Custom httpx client for making non-tracing HTTP requests. If not provided, a default client will be created.
 125        debug (bool): Enable debug logging. Defaults to False. Can also be set via LANGFUSE_DEBUG environment variable.
 126        tracing_enabled (Optional[bool]): Enable or disable tracing. Defaults to True. Can also be set via LANGFUSE_TRACING_ENABLED environment variable.
 127        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.
 128        flush_interval (Optional[float]): Time in seconds between batch flushes. Defaults to 5 seconds. Can also be set via LANGFUSE_FLUSH_INTERVAL environment variable.
 129        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'.
 130        release (Optional[str]): Release version/hash of your application. Used for grouping analytics by release.
 131        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.
 132        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.
 133        mask (Optional[MaskFunction]): Function to mask sensitive data in traces before sending to the API.
 134        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`)
 135        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.
 136        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.
 137
 138    Example:
 139        ```python
 140        from langfuse.otel import Langfuse
 141
 142        # Initialize the client (reads from env vars if not provided)
 143        langfuse = Langfuse(
 144            public_key="your-public-key",
 145            secret_key="your-secret-key",
 146            host="https://cloud.langfuse.com",  # Optional, default shown
 147        )
 148
 149        # Create a trace span
 150        with langfuse.start_as_current_span(name="process-query") as span:
 151            # Your application code here
 152
 153            # Create a nested generation span for an LLM call
 154            with span.start_as_current_generation(
 155                name="generate-response",
 156                model="gpt-4",
 157                input={"query": "Tell me about AI"},
 158                model_parameters={"temperature": 0.7, "max_tokens": 500}
 159            ) as generation:
 160                # Generate response here
 161                response = "AI is a field of computer science..."
 162
 163                generation.update(
 164                    output=response,
 165                    usage_details={"prompt_tokens": 10, "completion_tokens": 50},
 166                    cost_details={"total_cost": 0.0023}
 167                )
 168
 169                # Score the generation (supports NUMERIC, BOOLEAN, CATEGORICAL)
 170                generation.score(name="relevance", value=0.95, data_type="NUMERIC")
 171        ```
 172    """
 173
 174    _resources: Optional[LangfuseResourceManager] = None
 175    _mask: Optional[MaskFunction] = None
 176    _otel_tracer: otel_trace_api.Tracer
 177
 178    def __init__(
 179        self,
 180        *,
 181        public_key: Optional[str] = None,
 182        secret_key: Optional[str] = None,
 183        host: Optional[str] = None,
 184        timeout: Optional[int] = None,
 185        httpx_client: Optional[httpx.Client] = None,
 186        debug: bool = False,
 187        tracing_enabled: Optional[bool] = True,
 188        flush_at: Optional[int] = None,
 189        flush_interval: Optional[float] = None,
 190        environment: Optional[str] = None,
 191        release: Optional[str] = None,
 192        media_upload_thread_count: Optional[int] = None,
 193        sample_rate: Optional[float] = None,
 194        mask: Optional[MaskFunction] = None,
 195        blocked_instrumentation_scopes: Optional[List[str]] = None,
 196        additional_headers: Optional[Dict[str, str]] = None,
 197        tracer_provider: Optional[TracerProvider] = None,
 198    ):
 199        self._host = host or cast(
 200            str, os.environ.get(LANGFUSE_HOST, "https://cloud.langfuse.com")
 201        )
 202        self._environment = environment or cast(
 203            str, os.environ.get(LANGFUSE_TRACING_ENVIRONMENT)
 204        )
 205        self._project_id: Optional[str] = None
 206        sample_rate = sample_rate or float(os.environ.get(LANGFUSE_SAMPLE_RATE, 1.0))
 207        if not 0.0 <= sample_rate <= 1.0:
 208            raise ValueError(
 209                f"Sample rate must be between 0.0 and 1.0, got {sample_rate}"
 210            )
 211
 212        timeout = timeout or int(os.environ.get(LANGFUSE_TIMEOUT, 5))
 213
 214        self._tracing_enabled = (
 215            tracing_enabled
 216            and os.environ.get(LANGFUSE_TRACING_ENABLED, "true").lower() != "false"
 217        )
 218        if not self._tracing_enabled:
 219            langfuse_logger.info(
 220                "Configuration: Langfuse tracing is explicitly disabled. No data will be sent to the Langfuse API."
 221            )
 222
 223        debug = (
 224            debug if debug else (os.getenv(LANGFUSE_DEBUG, "false").lower() == "true")
 225        )
 226        if debug:
 227            logging.basicConfig(
 228                format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
 229            )
 230            langfuse_logger.setLevel(logging.DEBUG)
 231
 232        public_key = public_key or os.environ.get(LANGFUSE_PUBLIC_KEY)
 233        if public_key is None:
 234            langfuse_logger.warning(
 235                "Authentication error: Langfuse client initialized without public_key. Client will be disabled. "
 236                "Provide a public_key parameter or set LANGFUSE_PUBLIC_KEY environment variable. "
 237            )
 238            self._otel_tracer = otel_trace_api.NoOpTracer()
 239            return
 240
 241        secret_key = secret_key or os.environ.get(LANGFUSE_SECRET_KEY)
 242        if secret_key is None:
 243            langfuse_logger.warning(
 244                "Authentication error: Langfuse client initialized without secret_key. Client will be disabled. "
 245                "Provide a secret_key parameter or set LANGFUSE_SECRET_KEY environment variable. "
 246            )
 247            self._otel_tracer = otel_trace_api.NoOpTracer()
 248            return
 249
 250        if os.environ.get("OTEL_SDK_DISABLED", "false").lower() == "true":
 251            langfuse_logger.warning(
 252                "OTEL_SDK_DISABLED is set. Langfuse tracing will be disabled and no traces will appear in the UI."
 253            )
 254
 255        # Initialize api and tracer if requirements are met
 256        self._resources = LangfuseResourceManager(
 257            public_key=public_key,
 258            secret_key=secret_key,
 259            host=self._host,
 260            timeout=timeout,
 261            environment=environment,
 262            release=release,
 263            flush_at=flush_at,
 264            flush_interval=flush_interval,
 265            httpx_client=httpx_client,
 266            media_upload_thread_count=media_upload_thread_count,
 267            sample_rate=sample_rate,
 268            mask=mask,
 269            tracing_enabled=self._tracing_enabled,
 270            blocked_instrumentation_scopes=blocked_instrumentation_scopes,
 271            additional_headers=additional_headers,
 272            tracer_provider=tracer_provider,
 273        )
 274        self._mask = self._resources.mask
 275
 276        self._otel_tracer = (
 277            self._resources.tracer
 278            if self._tracing_enabled and self._resources.tracer is not None
 279            else otel_trace_api.NoOpTracer()
 280        )
 281        self.api = self._resources.api
 282        self.async_api = self._resources.async_api
 283
 284    def start_span(
 285        self,
 286        *,
 287        trace_context: Optional[TraceContext] = None,
 288        name: str,
 289        input: Optional[Any] = None,
 290        output: Optional[Any] = None,
 291        metadata: Optional[Any] = None,
 292        version: Optional[str] = None,
 293        level: Optional[SpanLevel] = None,
 294        status_message: Optional[str] = None,
 295    ) -> LangfuseSpan:
 296        """Create a new span for tracing a unit of work.
 297
 298        This method creates a new span but does not set it as the current span in the
 299        context. To create and use a span within a context, use start_as_current_span().
 300
 301        The created span will be the child of the current span in the context.
 302
 303        Args:
 304            trace_context: Optional context for connecting to an existing trace
 305            name: Name of the span (e.g., function or operation name)
 306            input: Input data for the operation (can be any JSON-serializable object)
 307            output: Output data from the operation (can be any JSON-serializable object)
 308            metadata: Additional metadata to associate with the span
 309            version: Version identifier for the code or component
 310            level: Importance level of the span (info, warning, error)
 311            status_message: Optional status message for the span
 312
 313        Returns:
 314            A LangfuseSpan object that must be ended with .end() when the operation completes
 315
 316        Example:
 317            ```python
 318            span = langfuse.start_span(name="process-data")
 319            try:
 320                # Do work
 321                span.update(output="result")
 322            finally:
 323                span.end()
 324            ```
 325        """
 326        return self.start_observation(
 327            trace_context=trace_context,
 328            name=name,
 329            as_type="span",
 330            input=input,
 331            output=output,
 332            metadata=metadata,
 333            version=version,
 334            level=level,
 335            status_message=status_message,
 336        )
 337
 338    def start_as_current_span(
 339        self,
 340        *,
 341        trace_context: Optional[TraceContext] = None,
 342        name: str,
 343        input: Optional[Any] = None,
 344        output: Optional[Any] = None,
 345        metadata: Optional[Any] = None,
 346        version: Optional[str] = None,
 347        level: Optional[SpanLevel] = None,
 348        status_message: Optional[str] = None,
 349        end_on_exit: Optional[bool] = None,
 350    ) -> _AgnosticContextManager[LangfuseSpan]:
 351        """Create a new span and set it as the current span in a context manager.
 352
 353        This method creates a new span and sets it as the current span within a context
 354        manager. Use this method with a 'with' statement to automatically handle span
 355        lifecycle within a code block.
 356
 357        The created span will be the child of the current span in the context.
 358
 359        Args:
 360            trace_context: Optional context for connecting to an existing trace
 361            name: Name of the span (e.g., function or operation name)
 362            input: Input data for the operation (can be any JSON-serializable object)
 363            output: Output data from the operation (can be any JSON-serializable object)
 364            metadata: Additional metadata to associate with the span
 365            version: Version identifier for the code or component
 366            level: Importance level of the span (info, warning, error)
 367            status_message: Optional status message for the span
 368            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.
 369
 370        Returns:
 371            A context manager that yields a LangfuseSpan
 372
 373        Example:
 374            ```python
 375            with langfuse.start_as_current_span(name="process-query") as span:
 376                # Do work
 377                result = process_data()
 378                span.update(output=result)
 379
 380                # Create a child span automatically
 381                with span.start_as_current_span(name="sub-operation") as child_span:
 382                    # Do sub-operation work
 383                    child_span.update(output="sub-result")
 384            ```
 385        """
 386        return self.start_as_current_observation(
 387            trace_context=trace_context,
 388            name=name,
 389            as_type="span",
 390            input=input,
 391            output=output,
 392            metadata=metadata,
 393            version=version,
 394            level=level,
 395            status_message=status_message,
 396            end_on_exit=end_on_exit,
 397        )
 398
 399    @overload
 400    def start_observation(
 401        self,
 402        *,
 403        trace_context: Optional[TraceContext] = None,
 404        name: str,
 405        as_type: Literal["generation"],
 406        input: Optional[Any] = None,
 407        output: Optional[Any] = None,
 408        metadata: Optional[Any] = None,
 409        version: Optional[str] = None,
 410        level: Optional[SpanLevel] = None,
 411        status_message: Optional[str] = None,
 412        completion_start_time: Optional[datetime] = None,
 413        model: Optional[str] = None,
 414        model_parameters: Optional[Dict[str, MapValue]] = None,
 415        usage_details: Optional[Dict[str, int]] = None,
 416        cost_details: Optional[Dict[str, float]] = None,
 417        prompt: Optional[PromptClient] = None,
 418    ) -> LangfuseGeneration: ...
 419
 420    @overload
 421    def start_observation(
 422        self,
 423        *,
 424        trace_context: Optional[TraceContext] = None,
 425        name: str,
 426        as_type: Literal["span"] = "span",
 427        input: Optional[Any] = None,
 428        output: Optional[Any] = None,
 429        metadata: Optional[Any] = None,
 430        version: Optional[str] = None,
 431        level: Optional[SpanLevel] = None,
 432        status_message: Optional[str] = None,
 433    ) -> LangfuseSpan: ...
 434
 435    @overload
 436    def start_observation(
 437        self,
 438        *,
 439        trace_context: Optional[TraceContext] = None,
 440        name: str,
 441        as_type: Literal["agent"],
 442        input: Optional[Any] = None,
 443        output: Optional[Any] = None,
 444        metadata: Optional[Any] = None,
 445        version: Optional[str] = None,
 446        level: Optional[SpanLevel] = None,
 447        status_message: Optional[str] = None,
 448    ) -> LangfuseAgent: ...
 449
 450    @overload
 451    def start_observation(
 452        self,
 453        *,
 454        trace_context: Optional[TraceContext] = None,
 455        name: str,
 456        as_type: Literal["tool"],
 457        input: Optional[Any] = None,
 458        output: Optional[Any] = None,
 459        metadata: Optional[Any] = None,
 460        version: Optional[str] = None,
 461        level: Optional[SpanLevel] = None,
 462        status_message: Optional[str] = None,
 463    ) -> LangfuseTool: ...
 464
 465    @overload
 466    def start_observation(
 467        self,
 468        *,
 469        trace_context: Optional[TraceContext] = None,
 470        name: str,
 471        as_type: Literal["chain"],
 472        input: Optional[Any] = None,
 473        output: Optional[Any] = None,
 474        metadata: Optional[Any] = None,
 475        version: Optional[str] = None,
 476        level: Optional[SpanLevel] = None,
 477        status_message: Optional[str] = None,
 478    ) -> LangfuseChain: ...
 479
 480    @overload
 481    def start_observation(
 482        self,
 483        *,
 484        trace_context: Optional[TraceContext] = None,
 485        name: str,
 486        as_type: Literal["retriever"],
 487        input: Optional[Any] = None,
 488        output: Optional[Any] = None,
 489        metadata: Optional[Any] = None,
 490        version: Optional[str] = None,
 491        level: Optional[SpanLevel] = None,
 492        status_message: Optional[str] = None,
 493    ) -> LangfuseRetriever: ...
 494
 495    @overload
 496    def start_observation(
 497        self,
 498        *,
 499        trace_context: Optional[TraceContext] = None,
 500        name: str,
 501        as_type: Literal["evaluator"],
 502        input: Optional[Any] = None,
 503        output: Optional[Any] = None,
 504        metadata: Optional[Any] = None,
 505        version: Optional[str] = None,
 506        level: Optional[SpanLevel] = None,
 507        status_message: Optional[str] = None,
 508    ) -> LangfuseEvaluator: ...
 509
 510    @overload
 511    def start_observation(
 512        self,
 513        *,
 514        trace_context: Optional[TraceContext] = None,
 515        name: str,
 516        as_type: Literal["embedding"],
 517        input: Optional[Any] = None,
 518        output: Optional[Any] = None,
 519        metadata: Optional[Any] = None,
 520        version: Optional[str] = None,
 521        level: Optional[SpanLevel] = None,
 522        status_message: Optional[str] = None,
 523        completion_start_time: Optional[datetime] = None,
 524        model: Optional[str] = None,
 525        model_parameters: Optional[Dict[str, MapValue]] = None,
 526        usage_details: Optional[Dict[str, int]] = None,
 527        cost_details: Optional[Dict[str, float]] = None,
 528        prompt: Optional[PromptClient] = None,
 529    ) -> LangfuseEmbedding: ...
 530
 531    @overload
 532    def start_observation(
 533        self,
 534        *,
 535        trace_context: Optional[TraceContext] = None,
 536        name: str,
 537        as_type: Literal["guardrail"],
 538        input: Optional[Any] = None,
 539        output: Optional[Any] = None,
 540        metadata: Optional[Any] = None,
 541        version: Optional[str] = None,
 542        level: Optional[SpanLevel] = None,
 543        status_message: Optional[str] = None,
 544    ) -> LangfuseGuardrail: ...
 545
 546    def start_observation(
 547        self,
 548        *,
 549        trace_context: Optional[TraceContext] = None,
 550        name: str,
 551        as_type: ObservationTypeLiteralNoEvent = "span",
 552        input: Optional[Any] = None,
 553        output: Optional[Any] = None,
 554        metadata: Optional[Any] = None,
 555        version: Optional[str] = None,
 556        level: Optional[SpanLevel] = None,
 557        status_message: Optional[str] = None,
 558        completion_start_time: Optional[datetime] = None,
 559        model: Optional[str] = None,
 560        model_parameters: Optional[Dict[str, MapValue]] = None,
 561        usage_details: Optional[Dict[str, int]] = None,
 562        cost_details: Optional[Dict[str, float]] = None,
 563        prompt: Optional[PromptClient] = None,
 564    ) -> Union[
 565        LangfuseSpan,
 566        LangfuseGeneration,
 567        LangfuseAgent,
 568        LangfuseTool,
 569        LangfuseChain,
 570        LangfuseRetriever,
 571        LangfuseEvaluator,
 572        LangfuseEmbedding,
 573        LangfuseGuardrail,
 574    ]:
 575        """Create a new observation of the specified type.
 576
 577        This method creates a new observation but does not set it as the current span in the
 578        context. To create and use an observation within a context, use start_as_current_observation().
 579
 580        Args:
 581            trace_context: Optional context for connecting to an existing trace
 582            name: Name of the observation
 583            as_type: Type of observation to create (defaults to "span")
 584            input: Input data for the operation
 585            output: Output data from the operation
 586            metadata: Additional metadata to associate with the observation
 587            version: Version identifier for the code or component
 588            level: Importance level of the observation
 589            status_message: Optional status message for the observation
 590            completion_start_time: When the model started generating (for generation types)
 591            model: Name/identifier of the AI model used (for generation types)
 592            model_parameters: Parameters used for the model (for generation types)
 593            usage_details: Token usage information (for generation types)
 594            cost_details: Cost information (for generation types)
 595            prompt: Associated prompt template (for generation types)
 596
 597        Returns:
 598            An observation object of the appropriate type that must be ended with .end()
 599        """
 600        if trace_context:
 601            trace_id = trace_context.get("trace_id", None)
 602            parent_span_id = trace_context.get("parent_span_id", None)
 603
 604            if trace_id:
 605                remote_parent_span = self._create_remote_parent_span(
 606                    trace_id=trace_id, parent_span_id=parent_span_id
 607                )
 608
 609                with otel_trace_api.use_span(
 610                    cast(otel_trace_api.Span, remote_parent_span)
 611                ):
 612                    otel_span = self._otel_tracer.start_span(name=name)
 613                    otel_span.set_attribute(LangfuseOtelSpanAttributes.AS_ROOT, True)
 614
 615                    return self._create_observation_from_otel_span(
 616                        otel_span=otel_span,
 617                        as_type=as_type,
 618                        input=input,
 619                        output=output,
 620                        metadata=metadata,
 621                        version=version,
 622                        level=level,
 623                        status_message=status_message,
 624                        completion_start_time=completion_start_time,
 625                        model=model,
 626                        model_parameters=model_parameters,
 627                        usage_details=usage_details,
 628                        cost_details=cost_details,
 629                        prompt=prompt,
 630                    )
 631
 632        otel_span = self._otel_tracer.start_span(name=name)
 633
 634        return self._create_observation_from_otel_span(
 635            otel_span=otel_span,
 636            as_type=as_type,
 637            input=input,
 638            output=output,
 639            metadata=metadata,
 640            version=version,
 641            level=level,
 642            status_message=status_message,
 643            completion_start_time=completion_start_time,
 644            model=model,
 645            model_parameters=model_parameters,
 646            usage_details=usage_details,
 647            cost_details=cost_details,
 648            prompt=prompt,
 649        )
 650
 651    def _create_observation_from_otel_span(
 652        self,
 653        *,
 654        otel_span: otel_trace_api.Span,
 655        as_type: ObservationTypeLiteralNoEvent,
 656        input: Optional[Any] = None,
 657        output: Optional[Any] = None,
 658        metadata: Optional[Any] = None,
 659        version: Optional[str] = None,
 660        level: Optional[SpanLevel] = None,
 661        status_message: Optional[str] = None,
 662        completion_start_time: Optional[datetime] = None,
 663        model: Optional[str] = None,
 664        model_parameters: Optional[Dict[str, MapValue]] = None,
 665        usage_details: Optional[Dict[str, int]] = None,
 666        cost_details: Optional[Dict[str, float]] = None,
 667        prompt: Optional[PromptClient] = None,
 668    ) -> Union[
 669        LangfuseSpan,
 670        LangfuseGeneration,
 671        LangfuseAgent,
 672        LangfuseTool,
 673        LangfuseChain,
 674        LangfuseRetriever,
 675        LangfuseEvaluator,
 676        LangfuseEmbedding,
 677        LangfuseGuardrail,
 678    ]:
 679        """Create the appropriate observation type from an OTEL span."""
 680        if as_type in get_observation_types_list(ObservationTypeGenerationLike):
 681            observation_class = self._get_span_class(as_type)
 682            # Type ignore to prevent overloads of internal _get_span_class function,
 683            # issue is that LangfuseEvent could be returned and that classes have diff. args
 684            return observation_class(  # type: ignore[return-value,call-arg]
 685                otel_span=otel_span,
 686                langfuse_client=self,
 687                environment=self._environment,
 688                input=input,
 689                output=output,
 690                metadata=metadata,
 691                version=version,
 692                level=level,
 693                status_message=status_message,
 694                completion_start_time=completion_start_time,
 695                model=model,
 696                model_parameters=model_parameters,
 697                usage_details=usage_details,
 698                cost_details=cost_details,
 699                prompt=prompt,
 700            )
 701        else:
 702            # For other types (e.g. span, guardrail), create appropriate class without generation properties
 703            observation_class = self._get_span_class(as_type)
 704            # Type ignore to prevent overloads of internal _get_span_class function,
 705            # issue is that LangfuseEvent could be returned and that classes have diff. args
 706            return observation_class(  # type: ignore[return-value,call-arg]
 707                otel_span=otel_span,
 708                langfuse_client=self,
 709                environment=self._environment,
 710                input=input,
 711                output=output,
 712                metadata=metadata,
 713                version=version,
 714                level=level,
 715                status_message=status_message,
 716            )
 717            # span._observation_type = as_type
 718            # span._otel_span.set_attribute("langfuse.observation.type", as_type)
 719            # return span
 720
 721    def start_generation(
 722        self,
 723        *,
 724        trace_context: Optional[TraceContext] = None,
 725        name: str,
 726        input: Optional[Any] = None,
 727        output: Optional[Any] = None,
 728        metadata: Optional[Any] = None,
 729        version: Optional[str] = None,
 730        level: Optional[SpanLevel] = None,
 731        status_message: Optional[str] = None,
 732        completion_start_time: Optional[datetime] = None,
 733        model: Optional[str] = None,
 734        model_parameters: Optional[Dict[str, MapValue]] = None,
 735        usage_details: Optional[Dict[str, int]] = None,
 736        cost_details: Optional[Dict[str, float]] = None,
 737        prompt: Optional[PromptClient] = None,
 738    ) -> LangfuseGeneration:
 739        """[DEPRECATED] Create a new generation span for model generations.
 740
 741        DEPRECATED: This method is deprecated and will be removed in a future version.
 742        Use start_observation(as_type='generation') instead.
 743
 744        This method creates a specialized span for tracking model generations.
 745        It includes additional fields specific to model generations such as model name,
 746        token usage, and cost details.
 747
 748        The created generation span will be the child of the current span in the context.
 749
 750        Args:
 751            trace_context: Optional context for connecting to an existing trace
 752            name: Name of the generation operation
 753            input: Input data for the model (e.g., prompts)
 754            output: Output from the model (e.g., completions)
 755            metadata: Additional metadata to associate with the generation
 756            version: Version identifier for the model or component
 757            level: Importance level of the generation (info, warning, error)
 758            status_message: Optional status message for the generation
 759            completion_start_time: When the model started generating the response
 760            model: Name/identifier of the AI model used (e.g., "gpt-4")
 761            model_parameters: Parameters used for the model (e.g., temperature, max_tokens)
 762            usage_details: Token usage information (e.g., prompt_tokens, completion_tokens)
 763            cost_details: Cost information for the model call
 764            prompt: Associated prompt template from Langfuse prompt management
 765
 766        Returns:
 767            A LangfuseGeneration object that must be ended with .end() when complete
 768
 769        Example:
 770            ```python
 771            generation = langfuse.start_generation(
 772                name="answer-generation",
 773                model="gpt-4",
 774                input={"prompt": "Explain quantum computing"},
 775                model_parameters={"temperature": 0.7}
 776            )
 777            try:
 778                # Call model API
 779                response = llm.generate(...)
 780
 781                generation.update(
 782                    output=response.text,
 783                    usage_details={
 784                        "prompt_tokens": response.usage.prompt_tokens,
 785                        "completion_tokens": response.usage.completion_tokens
 786                    }
 787                )
 788            finally:
 789                generation.end()
 790            ```
 791        """
 792        warnings.warn(
 793            "start_generation is deprecated and will be removed in a future version. "
 794            "Use start_observation(as_type='generation') instead.",
 795            DeprecationWarning,
 796            stacklevel=2,
 797        )
 798        return self.start_observation(
 799            trace_context=trace_context,
 800            name=name,
 801            as_type="generation",
 802            input=input,
 803            output=output,
 804            metadata=metadata,
 805            version=version,
 806            level=level,
 807            status_message=status_message,
 808            completion_start_time=completion_start_time,
 809            model=model,
 810            model_parameters=model_parameters,
 811            usage_details=usage_details,
 812            cost_details=cost_details,
 813            prompt=prompt,
 814        )
 815
 816    def start_as_current_generation(
 817        self,
 818        *,
 819        trace_context: Optional[TraceContext] = None,
 820        name: str,
 821        input: Optional[Any] = None,
 822        output: Optional[Any] = None,
 823        metadata: Optional[Any] = None,
 824        version: Optional[str] = None,
 825        level: Optional[SpanLevel] = None,
 826        status_message: Optional[str] = None,
 827        completion_start_time: Optional[datetime] = None,
 828        model: Optional[str] = None,
 829        model_parameters: Optional[Dict[str, MapValue]] = None,
 830        usage_details: Optional[Dict[str, int]] = None,
 831        cost_details: Optional[Dict[str, float]] = None,
 832        prompt: Optional[PromptClient] = None,
 833        end_on_exit: Optional[bool] = None,
 834    ) -> _AgnosticContextManager[LangfuseGeneration]:
 835        """[DEPRECATED] Create a new generation span and set it as the current span in a context manager.
 836
 837        DEPRECATED: This method is deprecated and will be removed in a future version.
 838        Use start_as_current_observation(as_type='generation') instead.
 839
 840        This method creates a specialized span for model generations and sets it as the
 841        current span within a context manager. Use this method with a 'with' statement to
 842        automatically handle the generation span lifecycle within a code block.
 843
 844        The created generation span will be the child of the current span in the context.
 845
 846        Args:
 847            trace_context: Optional context for connecting to an existing trace
 848            name: Name of the generation operation
 849            input: Input data for the model (e.g., prompts)
 850            output: Output from the model (e.g., completions)
 851            metadata: Additional metadata to associate with the generation
 852            version: Version identifier for the model or component
 853            level: Importance level of the generation (info, warning, error)
 854            status_message: Optional status message for the generation
 855            completion_start_time: When the model started generating the response
 856            model: Name/identifier of the AI model used (e.g., "gpt-4")
 857            model_parameters: Parameters used for the model (e.g., temperature, max_tokens)
 858            usage_details: Token usage information (e.g., prompt_tokens, completion_tokens)
 859            cost_details: Cost information for the model call
 860            prompt: Associated prompt template from Langfuse prompt management
 861            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.
 862
 863        Returns:
 864            A context manager that yields a LangfuseGeneration
 865
 866        Example:
 867            ```python
 868            with langfuse.start_as_current_generation(
 869                name="answer-generation",
 870                model="gpt-4",
 871                input={"prompt": "Explain quantum computing"}
 872            ) as generation:
 873                # Call model API
 874                response = llm.generate(...)
 875
 876                # Update with results
 877                generation.update(
 878                    output=response.text,
 879                    usage_details={
 880                        "prompt_tokens": response.usage.prompt_tokens,
 881                        "completion_tokens": response.usage.completion_tokens
 882                    }
 883                )
 884            ```
 885        """
 886        warnings.warn(
 887            "start_as_current_generation is deprecated and will be removed in a future version. "
 888            "Use start_as_current_observation(as_type='generation') instead.",
 889            DeprecationWarning,
 890            stacklevel=2,
 891        )
 892        return self.start_as_current_observation(
 893            trace_context=trace_context,
 894            name=name,
 895            as_type="generation",
 896            input=input,
 897            output=output,
 898            metadata=metadata,
 899            version=version,
 900            level=level,
 901            status_message=status_message,
 902            completion_start_time=completion_start_time,
 903            model=model,
 904            model_parameters=model_parameters,
 905            usage_details=usage_details,
 906            cost_details=cost_details,
 907            prompt=prompt,
 908            end_on_exit=end_on_exit,
 909        )
 910
 911    @overload
 912    def start_as_current_observation(
 913        self,
 914        *,
 915        trace_context: Optional[TraceContext] = None,
 916        name: str,
 917        as_type: Literal["generation"],
 918        input: Optional[Any] = None,
 919        output: Optional[Any] = None,
 920        metadata: Optional[Any] = None,
 921        version: Optional[str] = None,
 922        level: Optional[SpanLevel] = None,
 923        status_message: Optional[str] = None,
 924        completion_start_time: Optional[datetime] = None,
 925        model: Optional[str] = None,
 926        model_parameters: Optional[Dict[str, MapValue]] = None,
 927        usage_details: Optional[Dict[str, int]] = None,
 928        cost_details: Optional[Dict[str, float]] = None,
 929        prompt: Optional[PromptClient] = None,
 930        end_on_exit: Optional[bool] = None,
 931    ) -> _AgnosticContextManager[LangfuseGeneration]: ...
 932
 933    @overload
 934    def start_as_current_observation(
 935        self,
 936        *,
 937        trace_context: Optional[TraceContext] = None,
 938        name: str,
 939        as_type: Literal["span"] = "span",
 940        input: Optional[Any] = None,
 941        output: Optional[Any] = None,
 942        metadata: Optional[Any] = None,
 943        version: Optional[str] = None,
 944        level: Optional[SpanLevel] = None,
 945        status_message: Optional[str] = None,
 946        end_on_exit: Optional[bool] = None,
 947    ) -> _AgnosticContextManager[LangfuseSpan]: ...
 948
 949    @overload
 950    def start_as_current_observation(
 951        self,
 952        *,
 953        trace_context: Optional[TraceContext] = None,
 954        name: str,
 955        as_type: Literal["agent"],
 956        input: Optional[Any] = None,
 957        output: Optional[Any] = None,
 958        metadata: Optional[Any] = None,
 959        version: Optional[str] = None,
 960        level: Optional[SpanLevel] = None,
 961        status_message: Optional[str] = None,
 962        end_on_exit: Optional[bool] = None,
 963    ) -> _AgnosticContextManager[LangfuseAgent]: ...
 964
 965    @overload
 966    def start_as_current_observation(
 967        self,
 968        *,
 969        trace_context: Optional[TraceContext] = None,
 970        name: str,
 971        as_type: Literal["tool"],
 972        input: Optional[Any] = None,
 973        output: Optional[Any] = None,
 974        metadata: Optional[Any] = None,
 975        version: Optional[str] = None,
 976        level: Optional[SpanLevel] = None,
 977        status_message: Optional[str] = None,
 978        end_on_exit: Optional[bool] = None,
 979    ) -> _AgnosticContextManager[LangfuseTool]: ...
 980
 981    @overload
 982    def start_as_current_observation(
 983        self,
 984        *,
 985        trace_context: Optional[TraceContext] = None,
 986        name: str,
 987        as_type: Literal["chain"],
 988        input: Optional[Any] = None,
 989        output: Optional[Any] = None,
 990        metadata: Optional[Any] = None,
 991        version: Optional[str] = None,
 992        level: Optional[SpanLevel] = None,
 993        status_message: Optional[str] = None,
 994        end_on_exit: Optional[bool] = None,
 995    ) -> _AgnosticContextManager[LangfuseChain]: ...
 996
 997    @overload
 998    def start_as_current_observation(
 999        self,
1000        *,
1001        trace_context: Optional[TraceContext] = None,
1002        name: str,
1003        as_type: Literal["retriever"],
1004        input: Optional[Any] = None,
1005        output: Optional[Any] = None,
1006        metadata: Optional[Any] = None,
1007        version: Optional[str] = None,
1008        level: Optional[SpanLevel] = None,
1009        status_message: Optional[str] = None,
1010        end_on_exit: Optional[bool] = None,
1011    ) -> _AgnosticContextManager[LangfuseRetriever]: ...
1012
1013    @overload
1014    def start_as_current_observation(
1015        self,
1016        *,
1017        trace_context: Optional[TraceContext] = None,
1018        name: str,
1019        as_type: Literal["evaluator"],
1020        input: Optional[Any] = None,
1021        output: Optional[Any] = None,
1022        metadata: Optional[Any] = None,
1023        version: Optional[str] = None,
1024        level: Optional[SpanLevel] = None,
1025        status_message: Optional[str] = None,
1026        end_on_exit: Optional[bool] = None,
1027    ) -> _AgnosticContextManager[LangfuseEvaluator]: ...
1028
1029    @overload
1030    def start_as_current_observation(
1031        self,
1032        *,
1033        trace_context: Optional[TraceContext] = None,
1034        name: str,
1035        as_type: Literal["embedding"],
1036        input: Optional[Any] = None,
1037        output: Optional[Any] = None,
1038        metadata: Optional[Any] = None,
1039        version: Optional[str] = None,
1040        level: Optional[SpanLevel] = None,
1041        status_message: Optional[str] = None,
1042        completion_start_time: Optional[datetime] = None,
1043        model: Optional[str] = None,
1044        model_parameters: Optional[Dict[str, MapValue]] = None,
1045        usage_details: Optional[Dict[str, int]] = None,
1046        cost_details: Optional[Dict[str, float]] = None,
1047        prompt: Optional[PromptClient] = None,
1048        end_on_exit: Optional[bool] = None,
1049    ) -> _AgnosticContextManager[LangfuseEmbedding]: ...
1050
1051    @overload
1052    def start_as_current_observation(
1053        self,
1054        *,
1055        trace_context: Optional[TraceContext] = None,
1056        name: str,
1057        as_type: Literal["guardrail"],
1058        input: Optional[Any] = None,
1059        output: Optional[Any] = None,
1060        metadata: Optional[Any] = None,
1061        version: Optional[str] = None,
1062        level: Optional[SpanLevel] = None,
1063        status_message: Optional[str] = None,
1064        end_on_exit: Optional[bool] = None,
1065    ) -> _AgnosticContextManager[LangfuseGuardrail]: ...
1066
1067    def start_as_current_observation(
1068        self,
1069        *,
1070        trace_context: Optional[TraceContext] = None,
1071        name: str,
1072        as_type: ObservationTypeLiteralNoEvent = "span",
1073        input: Optional[Any] = None,
1074        output: Optional[Any] = None,
1075        metadata: Optional[Any] = None,
1076        version: Optional[str] = None,
1077        level: Optional[SpanLevel] = None,
1078        status_message: Optional[str] = None,
1079        completion_start_time: Optional[datetime] = None,
1080        model: Optional[str] = None,
1081        model_parameters: Optional[Dict[str, MapValue]] = None,
1082        usage_details: Optional[Dict[str, int]] = None,
1083        cost_details: Optional[Dict[str, float]] = None,
1084        prompt: Optional[PromptClient] = None,
1085        end_on_exit: Optional[bool] = None,
1086    ) -> Union[
1087        _AgnosticContextManager[LangfuseGeneration],
1088        _AgnosticContextManager[LangfuseSpan],
1089        _AgnosticContextManager[LangfuseAgent],
1090        _AgnosticContextManager[LangfuseTool],
1091        _AgnosticContextManager[LangfuseChain],
1092        _AgnosticContextManager[LangfuseRetriever],
1093        _AgnosticContextManager[LangfuseEvaluator],
1094        _AgnosticContextManager[LangfuseEmbedding],
1095        _AgnosticContextManager[LangfuseGuardrail],
1096    ]:
1097        """Create a new observation and set it as the current span in a context manager.
1098
1099        This method creates a new observation of the specified type and sets it as the
1100        current span within a context manager. Use this method with a 'with' statement to
1101        automatically handle the observation lifecycle within a code block.
1102
1103        The created observation will be the child of the current span in the context.
1104
1105        Args:
1106            trace_context: Optional context for connecting to an existing trace
1107            name: Name of the observation (e.g., function or operation name)
1108            as_type: Type of observation to create (defaults to "span")
1109            input: Input data for the operation (can be any JSON-serializable object)
1110            output: Output data from the operation (can be any JSON-serializable object)
1111            metadata: Additional metadata to associate with the observation
1112            version: Version identifier for the code or component
1113            level: Importance level of the observation (info, warning, error)
1114            status_message: Optional status message for the observation
1115            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.
1116
1117            The following parameters are available when as_type is: "generation" or "embedding".
1118            completion_start_time: When the model started generating the response
1119            model: Name/identifier of the AI model used (e.g., "gpt-4")
1120            model_parameters: Parameters used for the model (e.g., temperature, max_tokens)
1121            usage_details: Token usage information (e.g., prompt_tokens, completion_tokens)
1122            cost_details: Cost information for the model call
1123            prompt: Associated prompt template from Langfuse prompt management
1124
1125        Returns:
1126            A context manager that yields the appropriate observation type based on as_type
1127
1128        Example:
1129            ```python
1130            # Create a span
1131            with langfuse.start_as_current_observation(name="process-query", as_type="span") as span:
1132                # Do work
1133                result = process_data()
1134                span.update(output=result)
1135
1136                # Create a child span automatically
1137                with span.start_as_current_span(name="sub-operation") as child_span:
1138                    # Do sub-operation work
1139                    child_span.update(output="sub-result")
1140
1141            # Create a tool observation
1142            with langfuse.start_as_current_observation(name="web-search", as_type="tool") as tool:
1143                # Do tool work
1144                results = search_web(query)
1145                tool.update(output=results)
1146
1147            # Create a generation observation
1148            with langfuse.start_as_current_observation(
1149                name="answer-generation",
1150                as_type="generation",
1151                model="gpt-4"
1152            ) as generation:
1153                # Generate answer
1154                response = llm.generate(...)
1155                generation.update(output=response)
1156            ```
1157        """
1158        if as_type in get_observation_types_list(ObservationTypeGenerationLike):
1159            if trace_context:
1160                trace_id = trace_context.get("trace_id", None)
1161                parent_span_id = trace_context.get("parent_span_id", None)
1162
1163                if trace_id:
1164                    remote_parent_span = self._create_remote_parent_span(
1165                        trace_id=trace_id, parent_span_id=parent_span_id
1166                    )
1167
1168                    return cast(
1169                        Union[
1170                            _AgnosticContextManager[LangfuseGeneration],
1171                            _AgnosticContextManager[LangfuseEmbedding],
1172                        ],
1173                        self._create_span_with_parent_context(
1174                            as_type=as_type,
1175                            name=name,
1176                            remote_parent_span=remote_parent_span,
1177                            parent=None,
1178                            end_on_exit=end_on_exit,
1179                            input=input,
1180                            output=output,
1181                            metadata=metadata,
1182                            version=version,
1183                            level=level,
1184                            status_message=status_message,
1185                            completion_start_time=completion_start_time,
1186                            model=model,
1187                            model_parameters=model_parameters,
1188                            usage_details=usage_details,
1189                            cost_details=cost_details,
1190                            prompt=prompt,
1191                        ),
1192                    )
1193
1194            return cast(
1195                Union[
1196                    _AgnosticContextManager[LangfuseGeneration],
1197                    _AgnosticContextManager[LangfuseEmbedding],
1198                ],
1199                self._start_as_current_otel_span_with_processed_media(
1200                    as_type=as_type,
1201                    name=name,
1202                    end_on_exit=end_on_exit,
1203                    input=input,
1204                    output=output,
1205                    metadata=metadata,
1206                    version=version,
1207                    level=level,
1208                    status_message=status_message,
1209                    completion_start_time=completion_start_time,
1210                    model=model,
1211                    model_parameters=model_parameters,
1212                    usage_details=usage_details,
1213                    cost_details=cost_details,
1214                    prompt=prompt,
1215                ),
1216            )
1217
1218        if as_type in get_observation_types_list(ObservationTypeSpanLike):
1219            if trace_context:
1220                trace_id = trace_context.get("trace_id", None)
1221                parent_span_id = trace_context.get("parent_span_id", None)
1222
1223                if trace_id:
1224                    remote_parent_span = self._create_remote_parent_span(
1225                        trace_id=trace_id, parent_span_id=parent_span_id
1226                    )
1227
1228                    return cast(
1229                        Union[
1230                            _AgnosticContextManager[LangfuseSpan],
1231                            _AgnosticContextManager[LangfuseAgent],
1232                            _AgnosticContextManager[LangfuseTool],
1233                            _AgnosticContextManager[LangfuseChain],
1234                            _AgnosticContextManager[LangfuseRetriever],
1235                            _AgnosticContextManager[LangfuseEvaluator],
1236                            _AgnosticContextManager[LangfuseGuardrail],
1237                        ],
1238                        self._create_span_with_parent_context(
1239                            as_type=as_type,
1240                            name=name,
1241                            remote_parent_span=remote_parent_span,
1242                            parent=None,
1243                            end_on_exit=end_on_exit,
1244                            input=input,
1245                            output=output,
1246                            metadata=metadata,
1247                            version=version,
1248                            level=level,
1249                            status_message=status_message,
1250                        ),
1251                    )
1252
1253            return cast(
1254                Union[
1255                    _AgnosticContextManager[LangfuseSpan],
1256                    _AgnosticContextManager[LangfuseAgent],
1257                    _AgnosticContextManager[LangfuseTool],
1258                    _AgnosticContextManager[LangfuseChain],
1259                    _AgnosticContextManager[LangfuseRetriever],
1260                    _AgnosticContextManager[LangfuseEvaluator],
1261                    _AgnosticContextManager[LangfuseGuardrail],
1262                ],
1263                self._start_as_current_otel_span_with_processed_media(
1264                    as_type=as_type,
1265                    name=name,
1266                    end_on_exit=end_on_exit,
1267                    input=input,
1268                    output=output,
1269                    metadata=metadata,
1270                    version=version,
1271                    level=level,
1272                    status_message=status_message,
1273                ),
1274            )
1275
1276        # This should never be reached since all valid types are handled above
1277        langfuse_logger.warning(
1278            f"Unknown observation type: {as_type}, falling back to span"
1279        )
1280        return self._start_as_current_otel_span_with_processed_media(
1281            as_type="span",
1282            name=name,
1283            end_on_exit=end_on_exit,
1284            input=input,
1285            output=output,
1286            metadata=metadata,
1287            version=version,
1288            level=level,
1289            status_message=status_message,
1290        )
1291
1292    def _get_span_class(
1293        self,
1294        as_type: ObservationTypeLiteral,
1295    ) -> Union[
1296        Type[LangfuseAgent],
1297        Type[LangfuseTool],
1298        Type[LangfuseChain],
1299        Type[LangfuseRetriever],
1300        Type[LangfuseEvaluator],
1301        Type[LangfuseEmbedding],
1302        Type[LangfuseGuardrail],
1303        Type[LangfuseGeneration],
1304        Type[LangfuseEvent],
1305        Type[LangfuseSpan],
1306    ]:
1307        """Get the appropriate span class based on as_type."""
1308        normalized_type = as_type.lower()
1309
1310        if normalized_type == "agent":
1311            return LangfuseAgent
1312        elif normalized_type == "tool":
1313            return LangfuseTool
1314        elif normalized_type == "chain":
1315            return LangfuseChain
1316        elif normalized_type == "retriever":
1317            return LangfuseRetriever
1318        elif normalized_type == "evaluator":
1319            return LangfuseEvaluator
1320        elif normalized_type == "embedding":
1321            return LangfuseEmbedding
1322        elif normalized_type == "guardrail":
1323            return LangfuseGuardrail
1324        elif normalized_type == "generation":
1325            return LangfuseGeneration
1326        elif normalized_type == "event":
1327            return LangfuseEvent
1328        elif normalized_type == "span":
1329            return LangfuseSpan
1330        else:
1331            return LangfuseSpan
1332
1333    @_agnosticcontextmanager
1334    def _create_span_with_parent_context(
1335        self,
1336        *,
1337        name: str,
1338        parent: Optional[otel_trace_api.Span] = None,
1339        remote_parent_span: Optional[otel_trace_api.Span] = None,
1340        as_type: ObservationTypeLiteralNoEvent,
1341        end_on_exit: Optional[bool] = None,
1342        input: Optional[Any] = None,
1343        output: Optional[Any] = None,
1344        metadata: Optional[Any] = None,
1345        version: Optional[str] = None,
1346        level: Optional[SpanLevel] = None,
1347        status_message: Optional[str] = None,
1348        completion_start_time: Optional[datetime] = None,
1349        model: Optional[str] = None,
1350        model_parameters: Optional[Dict[str, MapValue]] = None,
1351        usage_details: Optional[Dict[str, int]] = None,
1352        cost_details: Optional[Dict[str, float]] = None,
1353        prompt: Optional[PromptClient] = None,
1354    ) -> Any:
1355        parent_span = parent or cast(otel_trace_api.Span, remote_parent_span)
1356
1357        with otel_trace_api.use_span(parent_span):
1358            with self._start_as_current_otel_span_with_processed_media(
1359                name=name,
1360                as_type=as_type,
1361                end_on_exit=end_on_exit,
1362                input=input,
1363                output=output,
1364                metadata=metadata,
1365                version=version,
1366                level=level,
1367                status_message=status_message,
1368                completion_start_time=completion_start_time,
1369                model=model,
1370                model_parameters=model_parameters,
1371                usage_details=usage_details,
1372                cost_details=cost_details,
1373                prompt=prompt,
1374            ) as langfuse_span:
1375                if remote_parent_span is not None:
1376                    langfuse_span._otel_span.set_attribute(
1377                        LangfuseOtelSpanAttributes.AS_ROOT, True
1378                    )
1379
1380                yield langfuse_span
1381
1382    @_agnosticcontextmanager
1383    def _start_as_current_otel_span_with_processed_media(
1384        self,
1385        *,
1386        name: str,
1387        as_type: Optional[ObservationTypeLiteralNoEvent] = None,
1388        end_on_exit: Optional[bool] = None,
1389        input: Optional[Any] = None,
1390        output: Optional[Any] = None,
1391        metadata: Optional[Any] = None,
1392        version: Optional[str] = None,
1393        level: Optional[SpanLevel] = None,
1394        status_message: Optional[str] = None,
1395        completion_start_time: Optional[datetime] = None,
1396        model: Optional[str] = None,
1397        model_parameters: Optional[Dict[str, MapValue]] = None,
1398        usage_details: Optional[Dict[str, int]] = None,
1399        cost_details: Optional[Dict[str, float]] = None,
1400        prompt: Optional[PromptClient] = None,
1401    ) -> Any:
1402        with self._otel_tracer.start_as_current_span(
1403            name=name,
1404            end_on_exit=end_on_exit if end_on_exit is not None else True,
1405        ) as otel_span:
1406            span_class = self._get_span_class(
1407                as_type or "generation"
1408            )  # default was "generation"
1409            common_args = {
1410                "otel_span": otel_span,
1411                "langfuse_client": self,
1412                "environment": self._environment,
1413                "input": input,
1414                "output": output,
1415                "metadata": metadata,
1416                "version": version,
1417                "level": level,
1418                "status_message": status_message,
1419            }
1420
1421            if span_class in [
1422                LangfuseGeneration,
1423                LangfuseEmbedding,
1424            ]:
1425                common_args.update(
1426                    {
1427                        "completion_start_time": completion_start_time,
1428                        "model": model,
1429                        "model_parameters": model_parameters,
1430                        "usage_details": usage_details,
1431                        "cost_details": cost_details,
1432                        "prompt": prompt,
1433                    }
1434                )
1435            # For span-like types (span, agent, tool, chain, retriever, evaluator, guardrail), no generation properties needed
1436
1437            yield span_class(**common_args)  # type: ignore[arg-type]
1438
1439    def _get_current_otel_span(self) -> Optional[otel_trace_api.Span]:
1440        current_span = otel_trace_api.get_current_span()
1441
1442        if current_span is otel_trace_api.INVALID_SPAN:
1443            langfuse_logger.warning(
1444                "Context error: No active span in current context. Operations that depend on an active span will be skipped. "
1445                "Ensure spans are created with start_as_current_span() or that you're operating within an active span context."
1446            )
1447            return None
1448
1449        return current_span
1450
1451    def update_current_generation(
1452        self,
1453        *,
1454        name: Optional[str] = None,
1455        input: Optional[Any] = None,
1456        output: Optional[Any] = None,
1457        metadata: Optional[Any] = None,
1458        version: Optional[str] = None,
1459        level: Optional[SpanLevel] = None,
1460        status_message: Optional[str] = None,
1461        completion_start_time: Optional[datetime] = None,
1462        model: Optional[str] = None,
1463        model_parameters: Optional[Dict[str, MapValue]] = None,
1464        usage_details: Optional[Dict[str, int]] = None,
1465        cost_details: Optional[Dict[str, float]] = None,
1466        prompt: Optional[PromptClient] = None,
1467    ) -> None:
1468        """Update the current active generation span with new information.
1469
1470        This method updates the current generation span in the active context with
1471        additional information. It's useful for adding output, usage stats, or other
1472        details that become available during or after model generation.
1473
1474        Args:
1475            name: The generation name
1476            input: Updated input data for the model
1477            output: Output from the model (e.g., completions)
1478            metadata: Additional metadata to associate with the generation
1479            version: Version identifier for the model or component
1480            level: Importance level of the generation (info, warning, error)
1481            status_message: Optional status message for the generation
1482            completion_start_time: When the model started generating the response
1483            model: Name/identifier of the AI model used (e.g., "gpt-4")
1484            model_parameters: Parameters used for the model (e.g., temperature, max_tokens)
1485            usage_details: Token usage information (e.g., prompt_tokens, completion_tokens)
1486            cost_details: Cost information for the model call
1487            prompt: Associated prompt template from Langfuse prompt management
1488
1489        Example:
1490            ```python
1491            with langfuse.start_as_current_generation(name="answer-query") as generation:
1492                # Initial setup and API call
1493                response = llm.generate(...)
1494
1495                # Update with results that weren't available at creation time
1496                langfuse.update_current_generation(
1497                    output=response.text,
1498                    usage_details={
1499                        "prompt_tokens": response.usage.prompt_tokens,
1500                        "completion_tokens": response.usage.completion_tokens
1501                    }
1502                )
1503            ```
1504        """
1505        if not self._tracing_enabled:
1506            langfuse_logger.debug(
1507                "Operation skipped: update_current_generation - Tracing is disabled or client is in no-op mode."
1508            )
1509            return
1510
1511        current_otel_span = self._get_current_otel_span()
1512
1513        if current_otel_span is not None:
1514            generation = LangfuseGeneration(
1515                otel_span=current_otel_span, langfuse_client=self
1516            )
1517
1518            if name:
1519                current_otel_span.update_name(name)
1520
1521            generation.update(
1522                input=input,
1523                output=output,
1524                metadata=metadata,
1525                version=version,
1526                level=level,
1527                status_message=status_message,
1528                completion_start_time=completion_start_time,
1529                model=model,
1530                model_parameters=model_parameters,
1531                usage_details=usage_details,
1532                cost_details=cost_details,
1533                prompt=prompt,
1534            )
1535
1536    def update_current_span(
1537        self,
1538        *,
1539        name: Optional[str] = None,
1540        input: Optional[Any] = None,
1541        output: Optional[Any] = None,
1542        metadata: Optional[Any] = None,
1543        version: Optional[str] = None,
1544        level: Optional[SpanLevel] = None,
1545        status_message: Optional[str] = None,
1546    ) -> None:
1547        """Update the current active span with new information.
1548
1549        This method updates the current span in the active context with
1550        additional information. It's useful for adding outputs or metadata
1551        that become available during execution.
1552
1553        Args:
1554            name: The span name
1555            input: Updated input data for the operation
1556            output: Output data from the operation
1557            metadata: Additional metadata to associate with the span
1558            version: Version identifier for the code or component
1559            level: Importance level of the span (info, warning, error)
1560            status_message: Optional status message for the span
1561
1562        Example:
1563            ```python
1564            with langfuse.start_as_current_span(name="process-data") as span:
1565                # Initial processing
1566                result = process_first_part()
1567
1568                # Update with intermediate results
1569                langfuse.update_current_span(metadata={"intermediate_result": result})
1570
1571                # Continue processing
1572                final_result = process_second_part(result)
1573
1574                # Final update
1575                langfuse.update_current_span(output=final_result)
1576            ```
1577        """
1578        if not self._tracing_enabled:
1579            langfuse_logger.debug(
1580                "Operation skipped: update_current_span - Tracing is disabled or client is in no-op mode."
1581            )
1582            return
1583
1584        current_otel_span = self._get_current_otel_span()
1585
1586        if current_otel_span is not None:
1587            span = LangfuseSpan(
1588                otel_span=current_otel_span,
1589                langfuse_client=self,
1590                environment=self._environment,
1591            )
1592
1593            if name:
1594                current_otel_span.update_name(name)
1595
1596            span.update(
1597                input=input,
1598                output=output,
1599                metadata=metadata,
1600                version=version,
1601                level=level,
1602                status_message=status_message,
1603            )
1604
1605    def update_current_trace(
1606        self,
1607        *,
1608        name: Optional[str] = None,
1609        user_id: Optional[str] = None,
1610        session_id: Optional[str] = None,
1611        version: Optional[str] = None,
1612        input: Optional[Any] = None,
1613        output: Optional[Any] = None,
1614        metadata: Optional[Any] = None,
1615        tags: Optional[List[str]] = None,
1616        public: Optional[bool] = None,
1617    ) -> None:
1618        """Update the current trace with additional information.
1619
1620        This method updates the Langfuse trace that the current span belongs to. It's useful for
1621        adding trace-level metadata like user ID, session ID, or tags that apply to
1622        the entire Langfuse trace rather than just a single observation.
1623
1624        Args:
1625            name: Updated name for the Langfuse trace
1626            user_id: ID of the user who initiated the Langfuse trace
1627            session_id: Session identifier for grouping related Langfuse traces
1628            version: Version identifier for the application or service
1629            input: Input data for the overall Langfuse trace
1630            output: Output data from the overall Langfuse trace
1631            metadata: Additional metadata to associate with the Langfuse trace
1632            tags: List of tags to categorize the Langfuse trace
1633            public: Whether the Langfuse trace should be publicly accessible
1634
1635        Example:
1636            ```python
1637            with langfuse.start_as_current_span(name="handle-request") as span:
1638                # Get user information
1639                user = authenticate_user(request)
1640
1641                # Update trace with user context
1642                langfuse.update_current_trace(
1643                    user_id=user.id,
1644                    session_id=request.session_id,
1645                    tags=["production", "web-app"]
1646                )
1647
1648                # Continue processing
1649                response = process_request(request)
1650
1651                # Update span with results
1652                span.update(output=response)
1653            ```
1654        """
1655        if not self._tracing_enabled:
1656            langfuse_logger.debug(
1657                "Operation skipped: update_current_trace - Tracing is disabled or client is in no-op mode."
1658            )
1659            return
1660
1661        current_otel_span = self._get_current_otel_span()
1662
1663        if current_otel_span is not None:
1664            existing_observation_type = current_otel_span.attributes.get(  # type: ignore[attr-defined]
1665                LangfuseOtelSpanAttributes.OBSERVATION_TYPE, "span"
1666            )
1667            # We need to preserve the class to keep the corret observation type
1668            span_class = self._get_span_class(existing_observation_type)
1669            span = span_class(
1670                otel_span=current_otel_span,
1671                langfuse_client=self,
1672                environment=self._environment,
1673            )
1674
1675            span.update_trace(
1676                name=name,
1677                user_id=user_id,
1678                session_id=session_id,
1679                version=version,
1680                input=input,
1681                output=output,
1682                metadata=metadata,
1683                tags=tags,
1684                public=public,
1685            )
1686
1687    def create_event(
1688        self,
1689        *,
1690        trace_context: Optional[TraceContext] = None,
1691        name: str,
1692        input: Optional[Any] = None,
1693        output: Optional[Any] = None,
1694        metadata: Optional[Any] = None,
1695        version: Optional[str] = None,
1696        level: Optional[SpanLevel] = None,
1697        status_message: Optional[str] = None,
1698    ) -> LangfuseEvent:
1699        """Create a new Langfuse observation of type 'EVENT'.
1700
1701        The created Langfuse Event observation will be the child of the current span in the context.
1702
1703        Args:
1704            trace_context: Optional context for connecting to an existing trace
1705            name: Name of the span (e.g., function or operation name)
1706            input: Input data for the operation (can be any JSON-serializable object)
1707            output: Output data from the operation (can be any JSON-serializable object)
1708            metadata: Additional metadata to associate with the span
1709            version: Version identifier for the code or component
1710            level: Importance level of the span (info, warning, error)
1711            status_message: Optional status message for the span
1712
1713        Returns:
1714            The Langfuse Event object
1715
1716        Example:
1717            ```python
1718            event = langfuse.create_event(name="process-event")
1719            ```
1720        """
1721        timestamp = time_ns()
1722
1723        if trace_context:
1724            trace_id = trace_context.get("trace_id", None)
1725            parent_span_id = trace_context.get("parent_span_id", None)
1726
1727            if trace_id:
1728                remote_parent_span = self._create_remote_parent_span(
1729                    trace_id=trace_id, parent_span_id=parent_span_id
1730                )
1731
1732                with otel_trace_api.use_span(
1733                    cast(otel_trace_api.Span, remote_parent_span)
1734                ):
1735                    otel_span = self._otel_tracer.start_span(
1736                        name=name, start_time=timestamp
1737                    )
1738                    otel_span.set_attribute(LangfuseOtelSpanAttributes.AS_ROOT, True)
1739
1740                    return cast(
1741                        LangfuseEvent,
1742                        LangfuseEvent(
1743                            otel_span=otel_span,
1744                            langfuse_client=self,
1745                            environment=self._environment,
1746                            input=input,
1747                            output=output,
1748                            metadata=metadata,
1749                            version=version,
1750                            level=level,
1751                            status_message=status_message,
1752                        ).end(end_time=timestamp),
1753                    )
1754
1755        otel_span = self._otel_tracer.start_span(name=name, start_time=timestamp)
1756
1757        return cast(
1758            LangfuseEvent,
1759            LangfuseEvent(
1760                otel_span=otel_span,
1761                langfuse_client=self,
1762                environment=self._environment,
1763                input=input,
1764                output=output,
1765                metadata=metadata,
1766                version=version,
1767                level=level,
1768                status_message=status_message,
1769            ).end(end_time=timestamp),
1770        )
1771
1772    def _create_remote_parent_span(
1773        self, *, trace_id: str, parent_span_id: Optional[str]
1774    ) -> Any:
1775        if not self._is_valid_trace_id(trace_id):
1776            langfuse_logger.warning(
1777                f"Passed trace ID '{trace_id}' is not a valid 32 lowercase hex char Langfuse trace id. Ignoring trace ID."
1778            )
1779
1780        if parent_span_id and not self._is_valid_span_id(parent_span_id):
1781            langfuse_logger.warning(
1782                f"Passed span ID '{parent_span_id}' is not a valid 16 lowercase hex char Langfuse span id. Ignoring parent span ID."
1783            )
1784
1785        int_trace_id = int(trace_id, 16)
1786        int_parent_span_id = (
1787            int(parent_span_id, 16)
1788            if parent_span_id
1789            else RandomIdGenerator().generate_span_id()
1790        )
1791
1792        span_context = otel_trace_api.SpanContext(
1793            trace_id=int_trace_id,
1794            span_id=int_parent_span_id,
1795            trace_flags=otel_trace_api.TraceFlags(0x01),  # mark span as sampled
1796            is_remote=False,
1797        )
1798
1799        return trace.NonRecordingSpan(span_context)
1800
1801    def _is_valid_trace_id(self, trace_id: str) -> bool:
1802        pattern = r"^[0-9a-f]{32}$"
1803
1804        return bool(re.match(pattern, trace_id))
1805
1806    def _is_valid_span_id(self, span_id: str) -> bool:
1807        pattern = r"^[0-9a-f]{16}$"
1808
1809        return bool(re.match(pattern, span_id))
1810
1811    def _create_observation_id(self, *, seed: Optional[str] = None) -> str:
1812        """Create a unique observation ID for use with Langfuse.
1813
1814        This method generates a unique observation ID (span ID in OpenTelemetry terms)
1815        for use with various Langfuse APIs. It can either generate a random ID or
1816        create a deterministic ID based on a seed string.
1817
1818        Observation IDs must be 16 lowercase hexadecimal characters, representing 8 bytes.
1819        This method ensures the generated ID meets this requirement. If you need to
1820        correlate an external ID with a Langfuse observation ID, use the external ID as
1821        the seed to get a valid, deterministic observation ID.
1822
1823        Args:
1824            seed: Optional string to use as a seed for deterministic ID generation.
1825                 If provided, the same seed will always produce the same ID.
1826                 If not provided, a random ID will be generated.
1827
1828        Returns:
1829            A 16-character lowercase hexadecimal string representing the observation ID.
1830
1831        Example:
1832            ```python
1833            # Generate a random observation ID
1834            obs_id = langfuse.create_observation_id()
1835
1836            # Generate a deterministic ID based on a seed
1837            user_obs_id = langfuse.create_observation_id(seed="user-123-feedback")
1838
1839            # Correlate an external item ID with a Langfuse observation ID
1840            item_id = "item-789012"
1841            correlated_obs_id = langfuse.create_observation_id(seed=item_id)
1842
1843            # Use the ID with Langfuse APIs
1844            langfuse.create_score(
1845                name="relevance",
1846                value=0.95,
1847                trace_id=trace_id,
1848                observation_id=obs_id
1849            )
1850            ```
1851        """
1852        if not seed:
1853            span_id_int = RandomIdGenerator().generate_span_id()
1854
1855            return self._format_otel_span_id(span_id_int)
1856
1857        return sha256(seed.encode("utf-8")).digest()[:8].hex()
1858
1859    @staticmethod
1860    def create_trace_id(*, seed: Optional[str] = None) -> str:
1861        """Create a unique trace ID for use with Langfuse.
1862
1863        This method generates a unique trace ID for use with various Langfuse APIs.
1864        It can either generate a random ID or create a deterministic ID based on
1865        a seed string.
1866
1867        Trace IDs must be 32 lowercase hexadecimal characters, representing 16 bytes.
1868        This method ensures the generated ID meets this requirement. If you need to
1869        correlate an external ID with a Langfuse trace ID, use the external ID as the
1870        seed to get a valid, deterministic Langfuse trace ID.
1871
1872        Args:
1873            seed: Optional string to use as a seed for deterministic ID generation.
1874                 If provided, the same seed will always produce the same ID.
1875                 If not provided, a random ID will be generated.
1876
1877        Returns:
1878            A 32-character lowercase hexadecimal string representing the Langfuse trace ID.
1879
1880        Example:
1881            ```python
1882            # Generate a random trace ID
1883            trace_id = langfuse.create_trace_id()
1884
1885            # Generate a deterministic ID based on a seed
1886            session_trace_id = langfuse.create_trace_id(seed="session-456")
1887
1888            # Correlate an external ID with a Langfuse trace ID
1889            external_id = "external-system-123456"
1890            correlated_trace_id = langfuse.create_trace_id(seed=external_id)
1891
1892            # Use the ID with trace context
1893            with langfuse.start_as_current_span(
1894                name="process-request",
1895                trace_context={"trace_id": trace_id}
1896            ) as span:
1897                # Operation will be part of the specific trace
1898                pass
1899            ```
1900        """
1901        if not seed:
1902            trace_id_int = RandomIdGenerator().generate_trace_id()
1903
1904            return Langfuse._format_otel_trace_id(trace_id_int)
1905
1906        return sha256(seed.encode("utf-8")).digest()[:16].hex()
1907
1908    def _get_otel_trace_id(self, otel_span: otel_trace_api.Span) -> str:
1909        span_context = otel_span.get_span_context()
1910
1911        return self._format_otel_trace_id(span_context.trace_id)
1912
1913    def _get_otel_span_id(self, otel_span: otel_trace_api.Span) -> str:
1914        span_context = otel_span.get_span_context()
1915
1916        return self._format_otel_span_id(span_context.span_id)
1917
1918    @staticmethod
1919    def _format_otel_span_id(span_id_int: int) -> str:
1920        """Format an integer span ID to a 16-character lowercase hex string.
1921
1922        Internal method to convert an OpenTelemetry integer span ID to the standard
1923        W3C Trace Context format (16-character lowercase hex string).
1924
1925        Args:
1926            span_id_int: 64-bit integer representing a span ID
1927
1928        Returns:
1929            A 16-character lowercase hexadecimal string
1930        """
1931        return format(span_id_int, "016x")
1932
1933    @staticmethod
1934    def _format_otel_trace_id(trace_id_int: int) -> str:
1935        """Format an integer trace ID to a 32-character lowercase hex string.
1936
1937        Internal method to convert an OpenTelemetry integer trace ID to the standard
1938        W3C Trace Context format (32-character lowercase hex string).
1939
1940        Args:
1941            trace_id_int: 128-bit integer representing a trace ID
1942
1943        Returns:
1944            A 32-character lowercase hexadecimal string
1945        """
1946        return format(trace_id_int, "032x")
1947
1948    @overload
1949    def create_score(
1950        self,
1951        *,
1952        name: str,
1953        value: float,
1954        session_id: Optional[str] = None,
1955        dataset_run_id: Optional[str] = None,
1956        trace_id: Optional[str] = None,
1957        observation_id: Optional[str] = None,
1958        score_id: Optional[str] = None,
1959        data_type: Optional[Literal["NUMERIC", "BOOLEAN"]] = None,
1960        comment: Optional[str] = None,
1961        config_id: Optional[str] = None,
1962        metadata: Optional[Any] = None,
1963    ) -> None: ...
1964
1965    @overload
1966    def create_score(
1967        self,
1968        *,
1969        name: str,
1970        value: str,
1971        session_id: Optional[str] = None,
1972        dataset_run_id: Optional[str] = None,
1973        trace_id: Optional[str] = None,
1974        score_id: Optional[str] = None,
1975        observation_id: Optional[str] = None,
1976        data_type: Optional[Literal["CATEGORICAL"]] = "CATEGORICAL",
1977        comment: Optional[str] = None,
1978        config_id: Optional[str] = None,
1979        metadata: Optional[Any] = None,
1980    ) -> None: ...
1981
1982    def create_score(
1983        self,
1984        *,
1985        name: str,
1986        value: Union[float, str],
1987        session_id: Optional[str] = None,
1988        dataset_run_id: Optional[str] = None,
1989        trace_id: Optional[str] = None,
1990        observation_id: Optional[str] = None,
1991        score_id: Optional[str] = None,
1992        data_type: Optional[ScoreDataType] = None,
1993        comment: Optional[str] = None,
1994        config_id: Optional[str] = None,
1995        metadata: Optional[Any] = None,
1996    ) -> None:
1997        """Create a score for a specific trace or observation.
1998
1999        This method creates a score for evaluating a Langfuse trace or observation. Scores can be
2000        used to track quality metrics, user feedback, or automated evaluations.
2001
2002        Args:
2003            name: Name of the score (e.g., "relevance", "accuracy")
2004            value: Score value (can be numeric for NUMERIC/BOOLEAN types or string for CATEGORICAL)
2005            session_id: ID of the Langfuse session to associate the score with
2006            dataset_run_id: ID of the Langfuse dataset run to associate the score with
2007            trace_id: ID of the Langfuse trace to associate the score with
2008            observation_id: Optional ID of the specific observation to score. Trace ID must be provided too.
2009            score_id: Optional custom ID for the score (auto-generated if not provided)
2010            data_type: Type of score (NUMERIC, BOOLEAN, or CATEGORICAL)
2011            comment: Optional comment or explanation for the score
2012            config_id: Optional ID of a score config defined in Langfuse
2013            metadata: Optional metadata to be attached to the score
2014
2015        Example:
2016            ```python
2017            # Create a numeric score for accuracy
2018            langfuse.create_score(
2019                name="accuracy",
2020                value=0.92,
2021                trace_id="abcdef1234567890abcdef1234567890",
2022                data_type="NUMERIC",
2023                comment="High accuracy with minor irrelevant details"
2024            )
2025
2026            # Create a categorical score for sentiment
2027            langfuse.create_score(
2028                name="sentiment",
2029                value="positive",
2030                trace_id="abcdef1234567890abcdef1234567890",
2031                observation_id="abcdef1234567890",
2032                data_type="CATEGORICAL"
2033            )
2034            ```
2035        """
2036        if not self._tracing_enabled:
2037            return
2038
2039        score_id = score_id or self._create_observation_id()
2040
2041        try:
2042            new_body = ScoreBody(
2043                id=score_id,
2044                sessionId=session_id,
2045                datasetRunId=dataset_run_id,
2046                traceId=trace_id,
2047                observationId=observation_id,
2048                name=name,
2049                value=value,
2050                dataType=data_type,  # type: ignore
2051                comment=comment,
2052                configId=config_id,
2053                environment=self._environment,
2054                metadata=metadata,
2055            )
2056
2057            event = {
2058                "id": self.create_trace_id(),
2059                "type": "score-create",
2060                "timestamp": _get_timestamp(),
2061                "body": new_body,
2062            }
2063
2064            if self._resources is not None:
2065                # Force the score to be in sample if it was for a legacy trace ID, i.e. non-32 hexchar
2066                force_sample = (
2067                    not self._is_valid_trace_id(trace_id) if trace_id else True
2068                )
2069
2070                self._resources.add_score_task(
2071                    event,
2072                    force_sample=force_sample,
2073                )
2074
2075        except Exception as e:
2076            langfuse_logger.exception(
2077                f"Error creating score: Failed to process score event for trace_id={trace_id}, name={name}. Error: {e}"
2078            )
2079
2080    @overload
2081    def score_current_span(
2082        self,
2083        *,
2084        name: str,
2085        value: float,
2086        score_id: Optional[str] = None,
2087        data_type: Optional[Literal["NUMERIC", "BOOLEAN"]] = None,
2088        comment: Optional[str] = None,
2089        config_id: Optional[str] = None,
2090    ) -> None: ...
2091
2092    @overload
2093    def score_current_span(
2094        self,
2095        *,
2096        name: str,
2097        value: str,
2098        score_id: Optional[str] = None,
2099        data_type: Optional[Literal["CATEGORICAL"]] = "CATEGORICAL",
2100        comment: Optional[str] = None,
2101        config_id: Optional[str] = None,
2102    ) -> None: ...
2103
2104    def score_current_span(
2105        self,
2106        *,
2107        name: str,
2108        value: Union[float, str],
2109        score_id: Optional[str] = None,
2110        data_type: Optional[ScoreDataType] = None,
2111        comment: Optional[str] = None,
2112        config_id: Optional[str] = None,
2113    ) -> None:
2114        """Create a score for the current active span.
2115
2116        This method scores the currently active span in the context. It's a convenient
2117        way to score the current operation without needing to know its trace and span IDs.
2118
2119        Args:
2120            name: Name of the score (e.g., "relevance", "accuracy")
2121            value: Score value (can be numeric for NUMERIC/BOOLEAN types or string for CATEGORICAL)
2122            score_id: Optional custom ID for the score (auto-generated if not provided)
2123            data_type: Type of score (NUMERIC, BOOLEAN, or CATEGORICAL)
2124            comment: Optional comment or explanation for the score
2125            config_id: Optional ID of a score config defined in Langfuse
2126
2127        Example:
2128            ```python
2129            with langfuse.start_as_current_generation(name="answer-query") as generation:
2130                # Generate answer
2131                response = generate_answer(...)
2132                generation.update(output=response)
2133
2134                # Score the generation
2135                langfuse.score_current_span(
2136                    name="relevance",
2137                    value=0.85,
2138                    data_type="NUMERIC",
2139                    comment="Mostly relevant but contains some tangential information"
2140                )
2141            ```
2142        """
2143        current_span = self._get_current_otel_span()
2144
2145        if current_span is not None:
2146            trace_id = self._get_otel_trace_id(current_span)
2147            observation_id = self._get_otel_span_id(current_span)
2148
2149            langfuse_logger.info(
2150                f"Score: Creating score name='{name}' value={value} for current span ({observation_id}) in trace {trace_id}"
2151            )
2152
2153            self.create_score(
2154                trace_id=trace_id,
2155                observation_id=observation_id,
2156                name=name,
2157                value=cast(str, value),
2158                score_id=score_id,
2159                data_type=cast(Literal["CATEGORICAL"], data_type),
2160                comment=comment,
2161                config_id=config_id,
2162            )
2163
2164    @overload
2165    def score_current_trace(
2166        self,
2167        *,
2168        name: str,
2169        value: float,
2170        score_id: Optional[str] = None,
2171        data_type: Optional[Literal["NUMERIC", "BOOLEAN"]] = None,
2172        comment: Optional[str] = None,
2173        config_id: Optional[str] = None,
2174    ) -> None: ...
2175
2176    @overload
2177    def score_current_trace(
2178        self,
2179        *,
2180        name: str,
2181        value: str,
2182        score_id: Optional[str] = None,
2183        data_type: Optional[Literal["CATEGORICAL"]] = "CATEGORICAL",
2184        comment: Optional[str] = None,
2185        config_id: Optional[str] = None,
2186    ) -> None: ...
2187
2188    def score_current_trace(
2189        self,
2190        *,
2191        name: str,
2192        value: Union[float, str],
2193        score_id: Optional[str] = None,
2194        data_type: Optional[ScoreDataType] = None,
2195        comment: Optional[str] = None,
2196        config_id: Optional[str] = None,
2197    ) -> None:
2198        """Create a score for the current trace.
2199
2200        This method scores the trace of the currently active span. Unlike score_current_span,
2201        this method associates the score with the entire trace rather than a specific span.
2202        It's useful for scoring overall performance or quality of the entire operation.
2203
2204        Args:
2205            name: Name of the score (e.g., "user_satisfaction", "overall_quality")
2206            value: Score value (can be numeric for NUMERIC/BOOLEAN types or string for CATEGORICAL)
2207            score_id: Optional custom ID for the score (auto-generated if not provided)
2208            data_type: Type of score (NUMERIC, BOOLEAN, or CATEGORICAL)
2209            comment: Optional comment or explanation for the score
2210            config_id: Optional ID of a score config defined in Langfuse
2211
2212        Example:
2213            ```python
2214            with langfuse.start_as_current_span(name="process-user-request") as span:
2215                # Process request
2216                result = process_complete_request()
2217                span.update(output=result)
2218
2219                # Score the overall trace
2220                langfuse.score_current_trace(
2221                    name="overall_quality",
2222                    value=0.95,
2223                    data_type="NUMERIC",
2224                    comment="High quality end-to-end response"
2225                )
2226            ```
2227        """
2228        current_span = self._get_current_otel_span()
2229
2230        if current_span is not None:
2231            trace_id = self._get_otel_trace_id(current_span)
2232
2233            langfuse_logger.info(
2234                f"Score: Creating score name='{name}' value={value} for entire trace {trace_id}"
2235            )
2236
2237            self.create_score(
2238                trace_id=trace_id,
2239                name=name,
2240                value=cast(str, value),
2241                score_id=score_id,
2242                data_type=cast(Literal["CATEGORICAL"], data_type),
2243                comment=comment,
2244                config_id=config_id,
2245            )
2246
2247    def flush(self) -> None:
2248        """Force flush all pending spans and events to the Langfuse API.
2249
2250        This method manually flushes any pending spans, scores, and other events to the
2251        Langfuse API. It's useful in scenarios where you want to ensure all data is sent
2252        before proceeding, without waiting for the automatic flush interval.
2253
2254        Example:
2255            ```python
2256            # Record some spans and scores
2257            with langfuse.start_as_current_span(name="operation") as span:
2258                # Do work...
2259                pass
2260
2261            # Ensure all data is sent to Langfuse before proceeding
2262            langfuse.flush()
2263
2264            # Continue with other work
2265            ```
2266        """
2267        if self._resources is not None:
2268            self._resources.flush()
2269
2270    def shutdown(self) -> None:
2271        """Shut down the Langfuse client and flush all pending data.
2272
2273        This method cleanly shuts down the Langfuse client, ensuring all pending data
2274        is flushed to the API and all background threads are properly terminated.
2275
2276        It's important to call this method when your application is shutting down to
2277        prevent data loss and resource leaks. For most applications, using the client
2278        as a context manager or relying on the automatic shutdown via atexit is sufficient.
2279
2280        Example:
2281            ```python
2282            # Initialize Langfuse
2283            langfuse = Langfuse(public_key="...", secret_key="...")
2284
2285            # Use Langfuse throughout your application
2286            # ...
2287
2288            # When application is shutting down
2289            langfuse.shutdown()
2290            ```
2291        """
2292        if self._resources is not None:
2293            self._resources.shutdown()
2294
2295    def get_current_trace_id(self) -> Optional[str]:
2296        """Get the trace ID of the current active span.
2297
2298        This method retrieves the trace ID from the currently active span in the context.
2299        It can be used to get the trace ID for referencing in logs, external systems,
2300        or for creating related operations.
2301
2302        Returns:
2303            The current trace ID as a 32-character lowercase hexadecimal string,
2304            or None if there is no active span.
2305
2306        Example:
2307            ```python
2308            with langfuse.start_as_current_span(name="process-request") as span:
2309                # Get the current trace ID for reference
2310                trace_id = langfuse.get_current_trace_id()
2311
2312                # Use it for external correlation
2313                log.info(f"Processing request with trace_id: {trace_id}")
2314
2315                # Or pass to another system
2316                external_system.process(data, trace_id=trace_id)
2317            ```
2318        """
2319        if not self._tracing_enabled:
2320            langfuse_logger.debug(
2321                "Operation skipped: get_current_trace_id - Tracing is disabled or client is in no-op mode."
2322            )
2323            return None
2324
2325        current_otel_span = self._get_current_otel_span()
2326
2327        return self._get_otel_trace_id(current_otel_span) if current_otel_span else None
2328
2329    def get_current_observation_id(self) -> Optional[str]:
2330        """Get the observation ID (span ID) of the current active span.
2331
2332        This method retrieves the observation ID from the currently active span in the context.
2333        It can be used to get the observation ID for referencing in logs, external systems,
2334        or for creating scores or other related operations.
2335
2336        Returns:
2337            The current observation ID as a 16-character lowercase hexadecimal string,
2338            or None if there is no active span.
2339
2340        Example:
2341            ```python
2342            with langfuse.start_as_current_span(name="process-user-query") as span:
2343                # Get the current observation ID
2344                observation_id = langfuse.get_current_observation_id()
2345
2346                # Store it for later reference
2347                cache.set(f"query_{query_id}_observation", observation_id)
2348
2349                # Process the query...
2350            ```
2351        """
2352        if not self._tracing_enabled:
2353            langfuse_logger.debug(
2354                "Operation skipped: get_current_observation_id - Tracing is disabled or client is in no-op mode."
2355            )
2356            return None
2357
2358        current_otel_span = self._get_current_otel_span()
2359
2360        return self._get_otel_span_id(current_otel_span) if current_otel_span else None
2361
2362    def _get_project_id(self) -> Optional[str]:
2363        """Fetch and return the current project id. Persisted across requests. Returns None if no project id is found for api keys."""
2364        if not self._project_id:
2365            proj = self.api.projects.get()
2366            if not proj.data or not proj.data[0].id:
2367                return None
2368
2369            self._project_id = proj.data[0].id
2370
2371        return self._project_id
2372
2373    def get_trace_url(self, *, trace_id: Optional[str] = None) -> Optional[str]:
2374        """Get the URL to view a trace in the Langfuse UI.
2375
2376        This method generates a URL that links directly to a trace in the Langfuse UI.
2377        It's useful for providing links in logs, notifications, or debugging tools.
2378
2379        Args:
2380            trace_id: Optional trace ID to generate a URL for. If not provided,
2381                     the trace ID of the current active span will be used.
2382
2383        Returns:
2384            A URL string pointing to the trace in the Langfuse UI,
2385            or None if the project ID couldn't be retrieved or no trace ID is available.
2386
2387        Example:
2388            ```python
2389            # Get URL for the current trace
2390            with langfuse.start_as_current_span(name="process-request") as span:
2391                trace_url = langfuse.get_trace_url()
2392                log.info(f"Processing trace: {trace_url}")
2393
2394            # Get URL for a specific trace
2395            specific_trace_url = langfuse.get_trace_url(trace_id="1234567890abcdef1234567890abcdef")
2396            send_notification(f"Review needed for trace: {specific_trace_url}")
2397            ```
2398        """
2399        project_id = self._get_project_id()
2400        final_trace_id = trace_id or self.get_current_trace_id()
2401
2402        return (
2403            f"{self._host}/project/{project_id}/traces/{final_trace_id}"
2404            if project_id and final_trace_id
2405            else None
2406        )
2407
2408    def get_dataset(
2409        self, name: str, *, fetch_items_page_size: Optional[int] = 50
2410    ) -> "DatasetClient":
2411        """Fetch a dataset by its name.
2412
2413        Args:
2414            name (str): The name of the dataset to fetch.
2415            fetch_items_page_size (Optional[int]): All items of the dataset will be fetched in chunks of this size. Defaults to 50.
2416
2417        Returns:
2418            DatasetClient: The dataset with the given name.
2419        """
2420        try:
2421            langfuse_logger.debug(f"Getting datasets {name}")
2422            dataset = self.api.datasets.get(dataset_name=name)
2423
2424            dataset_items = []
2425            page = 1
2426
2427            while True:
2428                new_items = self.api.dataset_items.list(
2429                    dataset_name=self._url_encode(name, is_url_param=True),
2430                    page=page,
2431                    limit=fetch_items_page_size,
2432                )
2433                dataset_items.extend(new_items.data)
2434
2435                if new_items.meta.total_pages <= page:
2436                    break
2437
2438                page += 1
2439
2440            items = [DatasetItemClient(i, langfuse=self) for i in dataset_items]
2441
2442            return DatasetClient(dataset, items=items)
2443
2444        except Error as e:
2445            handle_fern_exception(e)
2446            raise e
2447
2448    def auth_check(self) -> bool:
2449        """Check if the provided credentials (public and secret key) are valid.
2450
2451        Raises:
2452            Exception: If no projects were found for the provided credentials.
2453
2454        Note:
2455            This method is blocking. It is discouraged to use it in production code.
2456        """
2457        try:
2458            projects = self.api.projects.get()
2459            langfuse_logger.debug(
2460                f"Auth check successful, found {len(projects.data)} projects"
2461            )
2462            if len(projects.data) == 0:
2463                raise Exception(
2464                    "Auth check failed, no project found for the keys provided."
2465                )
2466            return True
2467
2468        except AttributeError as e:
2469            langfuse_logger.warning(
2470                f"Auth check failed: Client not properly initialized. Error: {e}"
2471            )
2472            return False
2473
2474        except Error as e:
2475            handle_fern_exception(e)
2476            raise e
2477
2478    def create_dataset(
2479        self,
2480        *,
2481        name: str,
2482        description: Optional[str] = None,
2483        metadata: Optional[Any] = None,
2484    ) -> Dataset:
2485        """Create a dataset with the given name on Langfuse.
2486
2487        Args:
2488            name: Name of the dataset to create.
2489            description: Description of the dataset. Defaults to None.
2490            metadata: Additional metadata. Defaults to None.
2491
2492        Returns:
2493            Dataset: The created dataset as returned by the Langfuse API.
2494        """
2495        try:
2496            body = CreateDatasetRequest(
2497                name=name, description=description, metadata=metadata
2498            )
2499            langfuse_logger.debug(f"Creating datasets {body}")
2500
2501            return self.api.datasets.create(request=body)
2502
2503        except Error as e:
2504            handle_fern_exception(e)
2505            raise e
2506
2507    def create_dataset_item(
2508        self,
2509        *,
2510        dataset_name: str,
2511        input: Optional[Any] = None,
2512        expected_output: Optional[Any] = None,
2513        metadata: Optional[Any] = None,
2514        source_trace_id: Optional[str] = None,
2515        source_observation_id: Optional[str] = None,
2516        status: Optional[DatasetStatus] = None,
2517        id: Optional[str] = None,
2518    ) -> DatasetItem:
2519        """Create a dataset item.
2520
2521        Upserts if an item with id already exists.
2522
2523        Args:
2524            dataset_name: Name of the dataset in which the dataset item should be created.
2525            input: Input data. Defaults to None. Can contain any dict, list or scalar.
2526            expected_output: Expected output data. Defaults to None. Can contain any dict, list or scalar.
2527            metadata: Additional metadata. Defaults to None. Can contain any dict, list or scalar.
2528            source_trace_id: Id of the source trace. Defaults to None.
2529            source_observation_id: Id of the source observation. Defaults to None.
2530            status: Status of the dataset item. Defaults to ACTIVE for newly created items.
2531            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.
2532
2533        Returns:
2534            DatasetItem: The created dataset item as returned by the Langfuse API.
2535
2536        Example:
2537            ```python
2538            from langfuse import Langfuse
2539
2540            langfuse = Langfuse()
2541
2542            # Uploading items to the Langfuse dataset named "capital_cities"
2543            langfuse.create_dataset_item(
2544                dataset_name="capital_cities",
2545                input={"input": {"country": "Italy"}},
2546                expected_output={"expected_output": "Rome"},
2547                metadata={"foo": "bar"}
2548            )
2549            ```
2550        """
2551        try:
2552            body = CreateDatasetItemRequest(
2553                datasetName=dataset_name,
2554                input=input,
2555                expectedOutput=expected_output,
2556                metadata=metadata,
2557                sourceTraceId=source_trace_id,
2558                sourceObservationId=source_observation_id,
2559                status=status,
2560                id=id,
2561            )
2562            langfuse_logger.debug(f"Creating dataset item {body}")
2563            return self.api.dataset_items.create(request=body)
2564        except Error as e:
2565            handle_fern_exception(e)
2566            raise e
2567
2568    def resolve_media_references(
2569        self,
2570        *,
2571        obj: Any,
2572        resolve_with: Literal["base64_data_uri"],
2573        max_depth: int = 10,
2574        content_fetch_timeout_seconds: int = 5,
2575    ) -> Any:
2576        """Replace media reference strings in an object with base64 data URIs.
2577
2578        This method recursively traverses an object (up to max_depth) looking for media reference strings
2579        in the format "@@@langfuseMedia:...@@@". When found, it (synchronously) fetches the actual media content using
2580        the provided Langfuse client and replaces the reference string with a base64 data URI.
2581
2582        If fetching media content fails for a reference string, a warning is logged and the reference
2583        string is left unchanged.
2584
2585        Args:
2586            obj: The object to process. Can be a primitive value, array, or nested object.
2587                If the object has a __dict__ attribute, a dict will be returned instead of the original object type.
2588            resolve_with: The representation of the media content to replace the media reference string with.
2589                Currently only "base64_data_uri" is supported.
2590            max_depth: int: The maximum depth to traverse the object. Default is 10.
2591            content_fetch_timeout_seconds: int: The timeout in seconds for fetching media content. Default is 5.
2592
2593        Returns:
2594            A deep copy of the input object with all media references replaced with base64 data URIs where possible.
2595            If the input object has a __dict__ attribute, a dict will be returned instead of the original object type.
2596
2597        Example:
2598            obj = {
2599                "image": "@@@langfuseMedia:type=image/jpeg|id=123|source=bytes@@@",
2600                "nested": {
2601                    "pdf": "@@@langfuseMedia:type=application/pdf|id=456|source=bytes@@@"
2602                }
2603            }
2604
2605            result = await LangfuseMedia.resolve_media_references(obj, langfuse_client)
2606
2607            # Result:
2608            # {
2609            #     "image": "data:image/jpeg;base64,/9j/4AAQSkZJRg...",
2610            #     "nested": {
2611            #         "pdf": "data:application/pdf;base64,JVBERi0xLjcK..."
2612            #     }
2613            # }
2614        """
2615        return LangfuseMedia.resolve_media_references(
2616            langfuse_client=self,
2617            obj=obj,
2618            resolve_with=resolve_with,
2619            max_depth=max_depth,
2620            content_fetch_timeout_seconds=content_fetch_timeout_seconds,
2621        )
2622
2623    @overload
2624    def get_prompt(
2625        self,
2626        name: str,
2627        *,
2628        version: Optional[int] = None,
2629        label: Optional[str] = None,
2630        type: Literal["chat"],
2631        cache_ttl_seconds: Optional[int] = None,
2632        fallback: Optional[List[ChatMessageDict]] = None,
2633        max_retries: Optional[int] = None,
2634        fetch_timeout_seconds: Optional[int] = None,
2635    ) -> ChatPromptClient: ...
2636
2637    @overload
2638    def get_prompt(
2639        self,
2640        name: str,
2641        *,
2642        version: Optional[int] = None,
2643        label: Optional[str] = None,
2644        type: Literal["text"] = "text",
2645        cache_ttl_seconds: Optional[int] = None,
2646        fallback: Optional[str] = None,
2647        max_retries: Optional[int] = None,
2648        fetch_timeout_seconds: Optional[int] = None,
2649    ) -> TextPromptClient: ...
2650
2651    def get_prompt(
2652        self,
2653        name: str,
2654        *,
2655        version: Optional[int] = None,
2656        label: Optional[str] = None,
2657        type: Literal["chat", "text"] = "text",
2658        cache_ttl_seconds: Optional[int] = None,
2659        fallback: Union[Optional[List[ChatMessageDict]], Optional[str]] = None,
2660        max_retries: Optional[int] = None,
2661        fetch_timeout_seconds: Optional[int] = None,
2662    ) -> PromptClient:
2663        """Get a prompt.
2664
2665        This method attempts to fetch the requested prompt from the local cache. If the prompt is not found
2666        in the cache or if the cached prompt has expired, it will try to fetch the prompt from the server again
2667        and update the cache. If fetching the new prompt fails, and there is an expired prompt in the cache, it will
2668        return the expired prompt as a fallback.
2669
2670        Args:
2671            name (str): The name of the prompt to retrieve.
2672
2673        Keyword Args:
2674            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.
2675            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.
2676            cache_ttl_seconds: Optional[int]: Time-to-live in seconds for caching the prompt. Must be specified as a
2677            keyword argument. If not set, defaults to 60 seconds. Disables caching if set to 0.
2678            type: Literal["chat", "text"]: The type of the prompt to retrieve. Defaults to "text".
2679            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.
2680            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.
2681            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.
2682
2683        Returns:
2684            The prompt object retrieved from the cache or directly fetched if not cached or expired of type
2685            - TextPromptClient, if type argument is 'text'.
2686            - ChatPromptClient, if type argument is 'chat'.
2687
2688        Raises:
2689            Exception: Propagates any exceptions raised during the fetching of a new prompt, unless there is an
2690            expired prompt in the cache, in which case it logs a warning and returns the expired prompt.
2691        """
2692        if self._resources is None:
2693            raise Error(
2694                "SDK is not correctly initalized. Check the init logs for more details."
2695            )
2696        if version is not None and label is not None:
2697            raise ValueError("Cannot specify both version and label at the same time.")
2698
2699        if not name:
2700            raise ValueError("Prompt name cannot be empty.")
2701
2702        cache_key = PromptCache.generate_cache_key(name, version=version, label=label)
2703        bounded_max_retries = self._get_bounded_max_retries(
2704            max_retries, default_max_retries=2, max_retries_upper_bound=4
2705        )
2706
2707        langfuse_logger.debug(f"Getting prompt '{cache_key}'")
2708        cached_prompt = self._resources.prompt_cache.get(cache_key)
2709
2710        if cached_prompt is None or cache_ttl_seconds == 0:
2711            langfuse_logger.debug(
2712                f"Prompt '{cache_key}' not found in cache or caching disabled."
2713            )
2714            try:
2715                return self._fetch_prompt_and_update_cache(
2716                    name,
2717                    version=version,
2718                    label=label,
2719                    ttl_seconds=cache_ttl_seconds,
2720                    max_retries=bounded_max_retries,
2721                    fetch_timeout_seconds=fetch_timeout_seconds,
2722                )
2723            except Exception as e:
2724                if fallback:
2725                    langfuse_logger.warning(
2726                        f"Returning fallback prompt for '{cache_key}' due to fetch error: {e}"
2727                    )
2728
2729                    fallback_client_args: Dict[str, Any] = {
2730                        "name": name,
2731                        "prompt": fallback,
2732                        "type": type,
2733                        "version": version or 0,
2734                        "config": {},
2735                        "labels": [label] if label else [],
2736                        "tags": [],
2737                    }
2738
2739                    if type == "text":
2740                        return TextPromptClient(
2741                            prompt=Prompt_Text(**fallback_client_args),
2742                            is_fallback=True,
2743                        )
2744
2745                    if type == "chat":
2746                        return ChatPromptClient(
2747                            prompt=Prompt_Chat(**fallback_client_args),
2748                            is_fallback=True,
2749                        )
2750
2751                raise e
2752
2753        if cached_prompt.is_expired():
2754            langfuse_logger.debug(f"Stale prompt '{cache_key}' found in cache.")
2755            try:
2756                # refresh prompt in background thread, refresh_prompt deduplicates tasks
2757                langfuse_logger.debug(f"Refreshing prompt '{cache_key}' in background.")
2758
2759                def refresh_task() -> None:
2760                    self._fetch_prompt_and_update_cache(
2761                        name,
2762                        version=version,
2763                        label=label,
2764                        ttl_seconds=cache_ttl_seconds,
2765                        max_retries=bounded_max_retries,
2766                        fetch_timeout_seconds=fetch_timeout_seconds,
2767                    )
2768
2769                self._resources.prompt_cache.add_refresh_prompt_task(
2770                    cache_key,
2771                    refresh_task,
2772                )
2773                langfuse_logger.debug(
2774                    f"Returning stale prompt '{cache_key}' from cache."
2775                )
2776                # return stale prompt
2777                return cached_prompt.value
2778
2779            except Exception as e:
2780                langfuse_logger.warning(
2781                    f"Error when refreshing cached prompt '{cache_key}', returning cached version. Error: {e}"
2782                )
2783                # creation of refresh prompt task failed, return stale prompt
2784                return cached_prompt.value
2785
2786        return cached_prompt.value
2787
2788    def _fetch_prompt_and_update_cache(
2789        self,
2790        name: str,
2791        *,
2792        version: Optional[int] = None,
2793        label: Optional[str] = None,
2794        ttl_seconds: Optional[int] = None,
2795        max_retries: int,
2796        fetch_timeout_seconds: Optional[int],
2797    ) -> PromptClient:
2798        cache_key = PromptCache.generate_cache_key(name, version=version, label=label)
2799        langfuse_logger.debug(f"Fetching prompt '{cache_key}' from server...")
2800
2801        try:
2802
2803            @backoff.on_exception(
2804                backoff.constant, Exception, max_tries=max_retries + 1, logger=None
2805            )
2806            def fetch_prompts() -> Any:
2807                return self.api.prompts.get(
2808                    self._url_encode(name),
2809                    version=version,
2810                    label=label,
2811                    request_options={
2812                        "timeout_in_seconds": fetch_timeout_seconds,
2813                    }
2814                    if fetch_timeout_seconds is not None
2815                    else None,
2816                )
2817
2818            prompt_response = fetch_prompts()
2819
2820            prompt: PromptClient
2821            if prompt_response.type == "chat":
2822                prompt = ChatPromptClient(prompt_response)
2823            else:
2824                prompt = TextPromptClient(prompt_response)
2825
2826            if self._resources is not None:
2827                self._resources.prompt_cache.set(cache_key, prompt, ttl_seconds)
2828
2829            return prompt
2830
2831        except Exception as e:
2832            langfuse_logger.error(
2833                f"Error while fetching prompt '{cache_key}': {str(e)}"
2834            )
2835            raise e
2836
2837    def _get_bounded_max_retries(
2838        self,
2839        max_retries: Optional[int],
2840        *,
2841        default_max_retries: int = 2,
2842        max_retries_upper_bound: int = 4,
2843    ) -> int:
2844        if max_retries is None:
2845            return default_max_retries
2846
2847        bounded_max_retries = min(
2848            max(max_retries, 0),
2849            max_retries_upper_bound,
2850        )
2851
2852        return bounded_max_retries
2853
2854    @overload
2855    def create_prompt(
2856        self,
2857        *,
2858        name: str,
2859        prompt: List[Union[ChatMessageDict, ChatMessageWithPlaceholdersDict]],
2860        labels: List[str] = [],
2861        tags: Optional[List[str]] = None,
2862        type: Optional[Literal["chat"]],
2863        config: Optional[Any] = None,
2864        commit_message: Optional[str] = None,
2865    ) -> ChatPromptClient: ...
2866
2867    @overload
2868    def create_prompt(
2869        self,
2870        *,
2871        name: str,
2872        prompt: str,
2873        labels: List[str] = [],
2874        tags: Optional[List[str]] = None,
2875        type: Optional[Literal["text"]] = "text",
2876        config: Optional[Any] = None,
2877        commit_message: Optional[str] = None,
2878    ) -> TextPromptClient: ...
2879
2880    def create_prompt(
2881        self,
2882        *,
2883        name: str,
2884        prompt: Union[
2885            str, List[Union[ChatMessageDict, ChatMessageWithPlaceholdersDict]]
2886        ],
2887        labels: List[str] = [],
2888        tags: Optional[List[str]] = None,
2889        type: Optional[Literal["chat", "text"]] = "text",
2890        config: Optional[Any] = None,
2891        commit_message: Optional[str] = None,
2892    ) -> PromptClient:
2893        """Create a new prompt in Langfuse.
2894
2895        Keyword Args:
2896            name : The name of the prompt to be created.
2897            prompt : The content of the prompt to be created.
2898            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.
2899            labels: The labels of the prompt. Defaults to None. To create a default-served prompt, add the 'production' label.
2900            tags: The tags of the prompt. Defaults to None. Will be applied to all versions of the prompt.
2901            config: Additional structured data to be saved with the prompt. Defaults to None.
2902            type: The type of the prompt to be created. "chat" vs. "text". Defaults to "text".
2903            commit_message: Optional string describing the change.
2904
2905        Returns:
2906            TextPromptClient: The prompt if type argument is 'text'.
2907            ChatPromptClient: The prompt if type argument is 'chat'.
2908        """
2909        try:
2910            langfuse_logger.debug(f"Creating prompt {name=}, {labels=}")
2911
2912            if type == "chat":
2913                if not isinstance(prompt, list):
2914                    raise ValueError(
2915                        "For 'chat' type, 'prompt' must be a list of chat messages with role and content attributes."
2916                    )
2917                request: Union[CreatePromptRequest_Chat, CreatePromptRequest_Text] = (
2918                    CreatePromptRequest_Chat(
2919                        name=name,
2920                        prompt=cast(Any, prompt),
2921                        labels=labels,
2922                        tags=tags,
2923                        config=config or {},
2924                        commitMessage=commit_message,
2925                        type="chat",
2926                    )
2927                )
2928                server_prompt = self.api.prompts.create(request=request)
2929
2930                if self._resources is not None:
2931                    self._resources.prompt_cache.invalidate(name)
2932
2933                return ChatPromptClient(prompt=cast(Prompt_Chat, server_prompt))
2934
2935            if not isinstance(prompt, str):
2936                raise ValueError("For 'text' type, 'prompt' must be a string.")
2937
2938            request = CreatePromptRequest_Text(
2939                name=name,
2940                prompt=prompt,
2941                labels=labels,
2942                tags=tags,
2943                config=config or {},
2944                commitMessage=commit_message,
2945                type="text",
2946            )
2947
2948            server_prompt = self.api.prompts.create(request=request)
2949
2950            if self._resources is not None:
2951                self._resources.prompt_cache.invalidate(name)
2952
2953            return TextPromptClient(prompt=cast(Prompt_Text, server_prompt))
2954
2955        except Error as e:
2956            handle_fern_exception(e)
2957            raise e
2958
2959    def update_prompt(
2960        self,
2961        *,
2962        name: str,
2963        version: int,
2964        new_labels: List[str] = [],
2965    ) -> Any:
2966        """Update an existing prompt version in Langfuse. The Langfuse SDK prompt cache is invalidated for all prompts witht he specified name.
2967
2968        Args:
2969            name (str): The name of the prompt to update.
2970            version (int): The version number of the prompt to update.
2971            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 [].
2972
2973        Returns:
2974            Prompt: The updated prompt from the Langfuse API.
2975
2976        """
2977        updated_prompt = self.api.prompt_version.update(
2978            name=name,
2979            version=version,
2980            new_labels=new_labels,
2981        )
2982
2983        if self._resources is not None:
2984            self._resources.prompt_cache.invalidate(name)
2985
2986        return updated_prompt
2987
2988    def _url_encode(self, url: str, *, is_url_param: Optional[bool] = False) -> str:
2989        # httpx ≥ 0.28 does its own WHATWG-compliant quoting (eg. encodes bare
2990        # “%”, “?”, “#”, “|”, … in query/path parts).  Re-quoting here would
2991        # double-encode, so we skip when the value is about to be sent straight
2992        # to httpx (`is_url_param=True`) and the installed version is ≥ 0.28.
2993        if is_url_param and Version(httpx.__version__) >= Version("0.28.0"):
2994            return url
2995
2996        # urllib.parse.quote does not escape slashes "/" by default; we need to add safe="" to force escaping
2997        # we need add safe="" to force escaping of slashes
2998        # This is necessary for prompts in prompt folders
2999        return urllib.parse.quote(url, safe="")