From 6478803c418bb6c1c569bf177e515a8e2e59f84a Mon Sep 17 00:00:00 2001 From: ShellWen Date: Wed, 22 Jul 2026 18:48:50 +0800 Subject: [PATCH] feat(tracing): allow explicit observation start times --- langfuse/_client/client.py | 17 +++++++++++++++-- langfuse/_client/span.py | 18 ++++++++++++++++-- tests/unit/test_otel.py | 31 +++++++++++++++++++++++++++++++ 3 files changed, 62 insertions(+), 4 deletions(-) diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index d3f699399..d001f4121 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -521,6 +521,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, completion_start_time: Optional[datetime] = None, model: Optional[str] = None, model_parameters: Optional[Dict[str, MapValue]] = None, @@ -542,6 +543,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, ) -> LangfuseSpan: ... @overload @@ -557,6 +559,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, ) -> LangfuseAgent: ... @overload @@ -572,6 +575,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, ) -> LangfuseTool: ... @overload @@ -587,6 +591,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, ) -> LangfuseChain: ... @overload @@ -602,6 +607,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, ) -> LangfuseRetriever: ... @overload @@ -617,6 +623,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, ) -> LangfuseEvaluator: ... @overload @@ -632,6 +639,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, completion_start_time: Optional[datetime] = None, model: Optional[str] = None, model_parameters: Optional[Dict[str, MapValue]] = None, @@ -653,6 +661,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, ) -> LangfuseGuardrail: ... def start_observation( @@ -667,6 +676,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, completion_start_time: Optional[datetime] = None, model: Optional[str] = None, model_parameters: Optional[Dict[str, MapValue]] = None, @@ -699,6 +709,7 @@ def start_observation( version: Version identifier for the code or component level: Importance level of the observation status_message: Optional status message for the observation + start_time: Optional explicit start time in nanoseconds since epoch completion_start_time: When the model started generating (for generation types) model: Name/identifier of the AI model used (for generation types) model_parameters: Parameters used for the model (for generation types) @@ -721,7 +732,9 @@ def start_observation( with otel_trace_api.use_span( cast(otel_trace_api.Span, remote_parent_span) ): - otel_span = self._otel_tracer.start_span(name=name) + otel_span = self._otel_tracer.start_span( + name=name, start_time=start_time + ) otel_span.set_attribute(LangfuseOtelSpanAttributes.AS_ROOT, True) return self._create_observation_from_otel_span( @@ -741,7 +754,7 @@ def start_observation( prompt=prompt, ) - otel_span = self._otel_tracer.start_span(name=name) + otel_span = self._otel_tracer.start_span(name=name, start_time=start_time) return self._create_observation_from_otel_span( otel_span=otel_span, diff --git a/langfuse/_client/span.py b/langfuse/_client/span.py index 71f8d68c7..d43e4e11c 100644 --- a/langfuse/_client/span.py +++ b/langfuse/_client/span.py @@ -745,6 +745,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, ) -> "LangfuseSpan": ... @overload @@ -759,6 +760,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, completion_start_time: Optional[datetime] = None, model: Optional[str] = None, model_parameters: Optional[Dict[str, MapValue]] = None, @@ -779,6 +781,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, ) -> "LangfuseAgent": ... @overload @@ -793,6 +796,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, ) -> "LangfuseTool": ... @overload @@ -807,6 +811,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, ) -> "LangfuseChain": ... @overload @@ -821,6 +826,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, ) -> "LangfuseRetriever": ... @overload @@ -835,6 +841,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, ) -> "LangfuseEvaluator": ... @overload @@ -849,6 +856,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, completion_start_time: Optional[datetime] = None, model: Optional[str] = None, model_parameters: Optional[Dict[str, MapValue]] = None, @@ -869,6 +877,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, ) -> "LangfuseGuardrail": ... @overload @@ -883,6 +892,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, ) -> "LangfuseEvent": ... def start_observation( @@ -896,6 +906,7 @@ def start_observation( version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, + start_time: Optional[int] = None, completion_start_time: Optional[datetime] = None, model: Optional[str] = None, model_parameters: Optional[Dict[str, MapValue]] = None, @@ -929,6 +940,7 @@ def start_observation( version: Version identifier for the code or component level: Importance level of the observation (info, warning, error) status_message: Optional status message for the observation + start_time: Optional explicit start time in nanoseconds since epoch completion_start_time: When the model started generating (for generation types) model: Name/identifier of the AI model used (for generation types) model_parameters: Parameters used for the model (for generation types) @@ -940,7 +952,7 @@ def start_observation( A new observation of the specified type that must be ended with .end() """ if as_type == "event": - timestamp = time_ns() + timestamp = start_time if start_time is not None else time_ns() event_span = self._langfuse_client._otel_tracer.start_span( name=name, start_time=timestamp ) @@ -968,7 +980,9 @@ def start_observation( observation_class = LangfuseSpan with otel_trace_api.use_span(self._otel_span): - new_otel_span = self._langfuse_client._otel_tracer.start_span(name=name) + new_otel_span = self._langfuse_client._otel_tracer.start_span( + name=name, start_time=start_time + ) common_args = { "otel_span": new_otel_span, diff --git a/tests/unit/test_otel.py b/tests/unit/test_otel.py index 46a085a71..7980ee8e8 100644 --- a/tests/unit/test_otel.py +++ b/tests/unit/test_otel.py @@ -3142,6 +3142,37 @@ def test_metrics_and_timing(self, langfuse_client, memory_exporter): f"Span duration ({span_duration_seconds}s) should be at least 0.05s" ) + def test_manual_observations_accept_historical_start_times( + self, langfuse_client, memory_exporter + ): + root_start_time = 1_750_000_000_000_000_000 + generation_start_time = root_start_time + 100_000_000 + end_time = root_start_time + 1_000_000_000 + + root = langfuse_client.start_observation( + name="historical-root", + trace_context={"trace_id": "a" * 32}, + start_time=root_start_time, + ) + generation = root.start_observation( + name="historical-generation", + as_type="generation", + start_time=generation_start_time, + ) + generation.end(end_time=end_time) + root.end(end_time=end_time) + + spans_by_name = { + span.name: span for span in memory_exporter.get_finished_spans() + } + root_span = spans_by_name["historical-root"] + generation_span = spans_by_name["historical-generation"] + + assert root_span.start_time == root_start_time + assert generation_span.start_time == generation_start_time + assert root_span.end_time == end_time + assert generation_span.end_time == end_time + # Add tests for media functionality in its own class class TestMediaHandling(TestOTelBase):