From 4506a58e1493269ab11caf3f8cf96a67e0ef6afc Mon Sep 17 00:00:00 2001 From: Hassieb Pakzad <68423100+hassiebp@users.noreply.github.com> Date: Fri, 10 Jul 2026 11:48:36 +0200 Subject: [PATCH 1/2] feat(openai): capture service_tier in model_parameters Capture the pricing-relevant service_tier attribute in the OpenAI integration: - Request side: extract the service_tier kwarg (guarding against NotGiven sentinels) and include it in model_parameters for chat completions and the Responses API. Embeddings are unchanged since they do not support service_tier. - Response side: read service_tier from the response in _get_langfuse_data_from_default_response and merge it into the request-side model_parameters on the generation update. The response value overrides the request value because OpenAI returns the tier that actually processed the request (e.g. when the request says "auto"), which is what pricing depends on. The merge preserves all request-side model parameters since span updates replace the whole model_parameters attribute. - Streaming: chat-completions streaming reads service_tier from the chunks; Responses API streaming reads it from the final response object. The request-side model_parameters are threaded through the stream wrappers so the finalized update can merge instead of clobber. Covers sync and async client paths, which share these helpers. Co-Authored-By: Claude Fable 5 --- langfuse/openai.py | 95 ++++++++++--- tests/unit/test_openai.py | 289 +++++++++++++++++++++++++++++++++++++- 2 files changed, 366 insertions(+), 18 deletions(-) diff --git a/langfuse/openai.py b/langfuse/openai.py index d78ed2850..ea9dcd68c 100644 --- a/langfuse/openai.py +++ b/langfuse/openai.py @@ -598,6 +598,12 @@ def _get_langfuse_data_from_kwargs(resource: OpenAiDefinition, kwargs: Any) -> A parsed_n = kwargs.get("n", 1) if not isinstance(kwargs.get("n", 1), NotGiven) else 1 + parsed_service_tier = ( + kwargs.get("service_tier", None) + if not isinstance(kwargs.get("service_tier", None), NotGiven) + else None + ) + if resource.type == "embedding": parsed_dimensions = ( kwargs.get("dimensions", None) @@ -634,6 +640,9 @@ def _get_langfuse_data_from_kwargs(resource: OpenAiDefinition, kwargs: Any) -> A if parsed_seed is not None: modelParameters["seed"] = parsed_seed + if parsed_service_tier is not None: + modelParameters["service_tier"] = parsed_service_tier + langfuse_prompt = kwargs.get("langfuse_prompt", None) return { @@ -657,6 +666,7 @@ def _create_langfuse_update( model: Optional[str] = None, usage: Optional[Any] = None, metadata: Optional[Any] = None, + model_parameters: Optional[Any] = None, ) -> Any: update = { "output": completion, @@ -668,6 +678,9 @@ def _create_langfuse_update( if metadata is not None: update["metadata"] = metadata + if model_parameters is not None: + update["model_parameters"] = model_parameters + if usage is not None: update["usage_details"] = _parse_usage(usage) update["cost_details"] = _parse_cost(usage) @@ -721,7 +734,7 @@ def _parse_cost(usage: Optional[Any] = None) -> Any: def _extract_streamed_response_api_response(chunks: Any) -> Any: - completion, model, usage = None, None, None + completion, model, usage, service_tier = None, None, None, None metadata = {} for raw_chunk in chunks: @@ -731,6 +744,7 @@ def _extract_streamed_response_api_response(chunks: Any) -> Any: response = raw_response.__dict__ model = response.get("model") + service_tier = response.get("service_tier", None) or service_tier for key, val in response.items(): if key not in ["created_at", "model", "output", "usage", "text"]: @@ -739,18 +753,19 @@ def _extract_streamed_response_api_response(chunks: Any) -> Any: if key == "output": completion = _extract_response_api_completion(val) - return (model, completion, usage, metadata) + return (model, completion, usage, metadata, service_tier) def _extract_streamed_openai_response(resource: Any, chunks: Any) -> Any: completion: Any = defaultdict(lambda: None) if resource.type == "chat" else "" - model, usage, finish_reason = None, None, None + model, usage, finish_reason, service_tier = None, None, None, None for chunk in chunks: if _is_openai_v1(): chunk = chunk.__dict__ model = model or chunk.get("model", None) or None + service_tier = service_tier or chunk.get("service_tier", None) or None chunk_usage = chunk.get("usage", None) if chunk_usage is not None: usage = chunk_usage @@ -884,6 +899,7 @@ def get_response_for_chat() -> Any: get_response_for_chat() if resource.type == "chat" else completion, usage, {"finish_reason": finish_reason} if finish_reason is not None else None, + service_tier, ) @@ -891,9 +907,10 @@ def _get_langfuse_data_from_default_response( resource: OpenAiDefinition, response: Any ) -> Any: if response is None: - return None, "", None + return None, "", None, None model = response.get("model", None) or None + service_tier = response.get("service_tier", None) or None completion = None @@ -942,7 +959,23 @@ def _get_langfuse_data_from_default_response( usage = _parse_usage(response.get("usage", None)) - return (model, completion, usage) + return (model, completion, usage, service_tier) + + +def _merge_service_tier_into_model_parameters( + model_parameters: Optional[Any], service_tier: Optional[Any] +) -> Optional[Any]: + """Merge the response-side service tier into the request-side model parameters. + + The response value is authoritative because OpenAI returns the tier that + actually processed the request (e.g. when the request specified "auto"). + Returns None when there is nothing to update so callers can skip the + update and keep the request-side model parameters untouched. + """ + if service_tier is None: + return None + + return {**(model_parameters or {}), "service_tier": service_tier} def _is_openai_v1() -> bool: @@ -999,9 +1032,10 @@ def _finalize_stream_response( items: list[Any], generation: LangfuseGeneration, completion_start_time: Optional[datetime], + model_parameters: Optional[Any] = None, ) -> None: try: - model, completion, usage, metadata = ( + model, completion, usage, metadata, service_tier = ( _extract_streamed_response_api_response(items) if resource.object == "Responses" or resource.object == "AsyncResponses" else _extract_streamed_openai_response(resource, items) @@ -1014,6 +1048,9 @@ def _finalize_stream_response( model=model, usage=usage, metadata=metadata, + model_parameters=_merge_service_tier_into_model_parameters( + model_parameters, service_tier + ), ) except Exception: pass @@ -1026,12 +1063,14 @@ def _instrument_openai_stream( resource: OpenAiDefinition, response: Any, generation: LangfuseGeneration, + model_parameters: Optional[Any] = None, ) -> Any: if not hasattr(response, "_iterator"): return LangfuseResponseGeneratorSync( resource=resource, response=response, generation=generation, + model_parameters=model_parameters, ) items: list[Any] = [] @@ -1051,6 +1090,7 @@ def finalize_once() -> None: items=items, generation=generation, completion_start_time=completion_start_time, + model_parameters=model_parameters, ) response._langfuse_finalize_once = finalize_once # type: ignore[attr-defined] @@ -1085,12 +1125,14 @@ def _instrument_openai_async_stream( resource: OpenAiDefinition, response: Any, generation: LangfuseGeneration, + model_parameters: Optional[Any] = None, ) -> Any: if not hasattr(response, "_iterator"): return LangfuseResponseGeneratorAsync( resource=resource, response=response, generation=generation, + model_parameters=model_parameters, ) items: list[Any] = [] @@ -1110,6 +1152,7 @@ async def finalize_once() -> None: items=items, generation=generation, completion_start_time=completion_start_time, + model_parameters=model_parameters, ) response._langfuse_finalize_once = finalize_once # type: ignore[attr-defined] @@ -1243,21 +1286,25 @@ def _wrap( resource=open_ai_resource, response=openai_response, generation=generation, + model_parameters=langfuse_data.get("model_parameters", None), ) elif _is_streaming_response(openai_response): return LangfuseResponseGeneratorSync( resource=open_ai_resource, response=openai_response, generation=generation, + model_parameters=langfuse_data.get("model_parameters", None), ) else: parsed_response = _unwrap_raw_response(openai_response) - model, completion, usage = _get_langfuse_data_from_default_response( - open_ai_resource, - (parsed_response and parsed_response.__dict__) - if _is_openai_v1() - else parsed_response, + model, completion, usage, service_tier = ( + _get_langfuse_data_from_default_response( + open_ai_resource, + (parsed_response and parsed_response.__dict__) + if _is_openai_v1() + else parsed_response, + ) ) generation.update( @@ -1267,6 +1314,9 @@ def _wrap( cost_details=_parse_cost(parsed_response.usage) if hasattr(parsed_response, "usage") else None, + model_parameters=_merge_service_tier_into_model_parameters( + langfuse_data.get("model_parameters", None), service_tier + ), ).end() return openai_response @@ -1325,21 +1375,25 @@ async def _wrap_async( resource=open_ai_resource, response=openai_response, generation=generation, + model_parameters=langfuse_data.get("model_parameters", None), ) elif _is_streaming_response(openai_response): return LangfuseResponseGeneratorAsync( resource=open_ai_resource, response=openai_response, generation=generation, + model_parameters=langfuse_data.get("model_parameters", None), ) else: parsed_response = _unwrap_raw_response(openai_response) - model, completion, usage = _get_langfuse_data_from_default_response( - open_ai_resource, - (parsed_response and parsed_response.__dict__) - if _is_openai_v1() - else parsed_response, + model, completion, usage, service_tier = ( + _get_langfuse_data_from_default_response( + open_ai_resource, + (parsed_response and parsed_response.__dict__) + if _is_openai_v1() + else parsed_response, + ) ) generation.update( model=model, @@ -1349,6 +1403,9 @@ async def _wrap_async( cost_details=_parse_cost(parsed_response.usage) if hasattr(parsed_response, "usage") else None, + model_parameters=_merge_service_tier_into_model_parameters( + langfuse_data.get("model_parameters", None), service_tier + ), ).end() return openai_response @@ -1397,12 +1454,14 @@ def __init__( resource: Any, response: Any, generation: Any, + model_parameters: Optional[Any] = None, ) -> None: self.items: list[Any] = [] self.resource = resource self.response = response self.generation = generation + self.model_parameters = model_parameters self.completion_start_time: Optional[datetime] = None self._is_finalized = False @@ -1458,6 +1517,7 @@ def _finalize(self) -> None: items=self.items, generation=self.generation, completion_start_time=self.completion_start_time, + model_parameters=self.model_parameters, ) @@ -1468,12 +1528,14 @@ def __init__( resource: Any, response: Any, generation: Any, + model_parameters: Optional[Any] = None, ) -> None: self.items: list[Any] = [] self.resource = resource self.response = response self.generation = generation + self.model_parameters = model_parameters self.completion_start_time: Optional[datetime] = None self._is_finalized = False @@ -1520,6 +1582,7 @@ async def _finalize(self) -> None: items=self.items, generation=self.generation, completion_start_time=self.completion_start_time, + model_parameters=self.model_parameters, ) async def close(self) -> None: diff --git a/tests/unit/test_openai.py b/tests/unit/test_openai.py index 78195e92e..681be4fbf 100644 --- a/tests/unit/test_openai.py +++ b/tests/unit/test_openai.py @@ -375,7 +375,7 @@ def test_openai_stream_with_none_choices_chunk_does_not_crash( def test_streaming_chat_completion_preserves_tool_calls_after_content(): - model, completion, usage, metadata = ( + model, completion, usage, metadata, _service_tier = ( lf_openai_module._extract_streamed_openai_response( SimpleNamespace(type="chat"), _make_chat_stream_chunks_with_content_before_tool_call(), @@ -405,7 +405,7 @@ def test_response_api_output_serializes_openai_parsed_response_objects(): class ParsedOutput(BaseModel): name: str - _, completion, _ = lf_openai_module._get_langfuse_data_from_default_response( + _, completion, _, _ = lf_openai_module._get_langfuse_data_from_default_response( SimpleNamespace(type="chat", object="Responses"), { "model": "gpt-4.1-mini", @@ -954,6 +954,291 @@ def test_embedding_exports_dimensions_and_count( } +def _make_chat_response(**extra_response_fields): + return SimpleNamespace( + model="gpt-4o-mini", + choices=[ + SimpleNamespace( + message=SimpleNamespace( + role="assistant", + content="2", + function_call=None, + tool_calls=None, + audio=None, + ) + ) + ], + usage=SimpleNamespace(prompt_tokens=3, completion_tokens=1, total_tokens=4), + **extra_response_fields, + ) + + +def test_chat_completion_captures_request_service_tier( + langfuse_memory_client, get_span, json_attr +): + openai_client = lf_openai.OpenAI(api_key="test") + response = _make_chat_response() + + with patch.object(openai_client.chat.completions, "_post", return_value=response): + openai_client.chat.completions.create( + name="unit-openai-service-tier-request", + model="gpt-4o-mini", + messages=[{"role": "user", "content": "1 + 1 = ?"}], + temperature=0, + service_tier="flex", + ) + + langfuse_memory_client.flush() + span = get_span("unit-openai-service-tier-request") + + model_parameters = json_attr( + span, LangfuseOtelSpanAttributes.OBSERVATION_MODEL_PARAMETERS + ) + assert model_parameters["service_tier"] == "flex" + assert model_parameters["temperature"] == 0 + + +def test_chat_completion_service_tier_not_given_is_absent( + langfuse_memory_client, get_span, json_attr +): + from openai._types import NOT_GIVEN + + openai_client = lf_openai.OpenAI(api_key="test") + response = _make_chat_response() + + with patch.object(openai_client.chat.completions, "_post", return_value=response): + openai_client.chat.completions.create( + name="unit-openai-service-tier-not-given", + model="gpt-4o-mini", + messages=[{"role": "user", "content": "1 + 1 = ?"}], + temperature=0, + service_tier=NOT_GIVEN, + ) + + langfuse_memory_client.flush() + span = get_span("unit-openai-service-tier-not-given") + + model_parameters = json_attr( + span, LangfuseOtelSpanAttributes.OBSERVATION_MODEL_PARAMETERS + ) + assert "service_tier" not in model_parameters + + +def test_chat_completion_service_tier_absent_by_default( + langfuse_memory_client, get_span, json_attr +): + openai_client = lf_openai.OpenAI(api_key="test") + response = _make_chat_response() + + with patch.object(openai_client.chat.completions, "_post", return_value=response): + openai_client.chat.completions.create( + name="unit-openai-service-tier-absent", + model="gpt-4o-mini", + messages=[{"role": "user", "content": "1 + 1 = ?"}], + temperature=0, + ) + + langfuse_memory_client.flush() + span = get_span("unit-openai-service-tier-absent") + + model_parameters = json_attr( + span, LangfuseOtelSpanAttributes.OBSERVATION_MODEL_PARAMETERS + ) + assert "service_tier" not in model_parameters + + +def test_chat_completion_response_service_tier_overrides_request( + langfuse_memory_client, get_span, json_attr +): + openai_client = lf_openai.OpenAI(api_key="test") + response = _make_chat_response(service_tier="default") + + with patch.object(openai_client.chat.completions, "_post", return_value=response): + openai_client.chat.completions.create( + name="unit-openai-service-tier-override", + model="gpt-4o-mini", + messages=[{"role": "user", "content": "1 + 1 = ?"}], + temperature=0, + service_tier="auto", + ) + + langfuse_memory_client.flush() + span = get_span("unit-openai-service-tier-override") + + model_parameters = json_attr( + span, LangfuseOtelSpanAttributes.OBSERVATION_MODEL_PARAMETERS + ) + # Response value is authoritative: it reflects the tier actually used. + assert model_parameters["service_tier"] == "default" + # Request-side model parameters must be preserved (merge, not clobber). + assert model_parameters["temperature"] == 0 + assert model_parameters["top_p"] == 1 + + +@pytest.mark.asyncio +async def test_async_chat_completion_response_service_tier_overrides_request( + langfuse_memory_client, get_span, json_attr +): + openai_client = lf_openai.AsyncOpenAI(api_key="test") + response = _make_chat_response(service_tier="priority") + + with patch.object(openai_client.chat.completions, "_post", return_value=response): + await openai_client.chat.completions.create( + name="unit-openai-async-service-tier-override", + model="gpt-4o-mini", + messages=[{"role": "user", "content": "1 + 1 = ?"}], + temperature=0, + service_tier="auto", + ) + + langfuse_memory_client.flush() + span = get_span("unit-openai-async-service-tier-override") + + model_parameters = json_attr( + span, LangfuseOtelSpanAttributes.OBSERVATION_MODEL_PARAMETERS + ) + assert model_parameters["service_tier"] == "priority" + assert model_parameters["temperature"] == 0 + + +def test_openai_stream_captures_service_tier_from_chunks( + langfuse_memory_client, get_span, json_attr +): + openai_client = lf_openai.OpenAI(api_key="test") + chunks = _make_chat_stream_chunks() + for chunk in chunks: + chunk.service_tier = "default" + raw_stream = DummyOpenAIStream(chunks, DummySyncResponse()) + + with patch.object(openai_client.chat.completions, "_post", return_value=raw_stream): + stream = openai_client.chat.completions.create( + name="unit-openai-stream-service-tier", + model="gpt-4o-mini", + messages=[{"role": "user", "content": "1 + 1 = ?"}], + temperature=0, + service_tier="auto", + stream=True, + ) + + list(stream) + stream.close() + + langfuse_memory_client.flush() + span = get_span("unit-openai-stream-service-tier") + + model_parameters = json_attr( + span, LangfuseOtelSpanAttributes.OBSERVATION_MODEL_PARAMETERS + ) + assert model_parameters["service_tier"] == "default" + assert model_parameters["temperature"] == 0 + + +@pytest.mark.asyncio +async def test_openai_async_stream_captures_service_tier_from_chunks( + langfuse_memory_client, get_span, json_attr +): + openai_client = lf_openai.AsyncOpenAI(api_key="test") + chunks = _make_chat_stream_chunks() + for chunk in chunks: + chunk.service_tier = "flex" + raw_stream = DummyOpenAIAsyncStream(chunks, DummyAsyncResponse()) + + with patch.object(openai_client.chat.completions, "_post", return_value=raw_stream): + stream = await openai_client.chat.completions.create( + name="unit-openai-async-stream-service-tier", + model="gpt-4o-mini", + messages=[{"role": "user", "content": "1 + 1 = ?"}], + temperature=0, + stream=True, + ) + + async for _ in stream: + pass + + await stream.aclose() + + langfuse_memory_client.flush() + span = get_span("unit-openai-async-stream-service-tier") + + model_parameters = json_attr( + span, LangfuseOtelSpanAttributes.OBSERVATION_MODEL_PARAMETERS + ) + assert model_parameters["service_tier"] == "flex" + assert model_parameters["temperature"] == 0 + + +def test_embedding_model_parameters_do_not_include_service_tier( + langfuse_memory_client, get_span, json_attr +): + openai_client = lf_openai.OpenAI(api_key="test") + response = SimpleNamespace( + model="text-embedding-3-small", + data=[SimpleNamespace(embedding=[0.1, 0.2, 0.3])], + usage=SimpleNamespace(prompt_tokens=2, total_tokens=2), + ) + + with patch.object(openai_client.embeddings, "_post", return_value=response): + openai_client.embeddings.create( + name="unit-openai-embedding-no-service-tier", + model="text-embedding-3-small", + input="hello world", + dimensions=3, + ) + + langfuse_memory_client.flush() + span = get_span("unit-openai-embedding-no-service-tier") + + model_parameters = json_attr( + span, LangfuseOtelSpanAttributes.OBSERVATION_MODEL_PARAMETERS + ) + assert model_parameters == {"dimensions": 3} + + +def test_default_response_extraction_returns_service_tier(): + ( + model, + _completion, + _usage, + service_tier, + ) = lf_openai_module._get_langfuse_data_from_default_response( + SimpleNamespace(type="chat", object="Responses"), + { + "model": "gpt-4.1-mini", + "output": [], + "usage": None, + "service_tier": "flex", + }, + ) + + assert model == "gpt-4.1-mini" + assert service_tier == "flex" + + +def test_streamed_response_api_extraction_returns_service_tier(): + final_response = SimpleNamespace( + model="gpt-4.1-mini", + output=[], + usage=SimpleNamespace(input_tokens=1, output_tokens=1, total_tokens=2), + service_tier="priority", + created_at=1700000000, + text=None, + ) + chunks = [ + SimpleNamespace(type="response.completed", response=final_response), + ] + + ( + model, + _completion, + _usage, + _metadata, + service_tier, + ) = lf_openai_module._extract_streamed_response_api_response(chunks) + + assert model == "gpt-4.1-mini" + assert service_tier == "priority" + + def _chat_completion_payload(): return { "id": "chatcmpl-test", From e8c7696f791cd20d4c9db99d9eed168182a372e7 Mon Sep 17 00:00:00 2001 From: Hassieb Pakzad <68423100+hassiebp@users.noreply.github.com> Date: Fri, 10 Jul 2026 13:26:55 +0200 Subject: [PATCH 2/2] push --- tests/e2e/test_decorators.py | 1 + tests/live_provider/test_openai.py | 9 +++++++++ 2 files changed, 10 insertions(+) diff --git a/tests/e2e/test_decorators.py b/tests/e2e/test_decorators.py index 7c289980d..9e302f799 100644 --- a/tests/e2e/test_decorators.py +++ b/tests/e2e/test_decorators.py @@ -940,6 +940,7 @@ async def level_1_function(*args, **kwargs): assert generation.end_time is not None assert generation.start_time < generation.end_time assert generation.model_parameters == { + "service_tier": "default", "temperature": 0, "top_p": 1, "frequency_penalty": 0, diff --git a/tests/live_provider/test_openai.py b/tests/live_provider/test_openai.py index eb750970b..b0294fecd 100644 --- a/tests/live_provider/test_openai.py +++ b/tests/live_provider/test_openai.py @@ -74,6 +74,7 @@ def test_openai_chat_completion(openai): assert generation.data[0].end_time is not None assert generation.data[0].start_time < generation.data[0].end_time assert generation.data[0].model_parameters == { + "service_tier": "default", "temperature": 0, "top_p": 1, "frequency_penalty": 0, @@ -125,6 +126,7 @@ def test_openai_chat_completion_stream(openai): assert generation.data[0].end_time is not None assert generation.data[0].start_time < generation.data[0].end_time assert generation.data[0].model_parameters == { + "service_tier": "default", "temperature": 0, "top_p": 1, "frequency_penalty": 0, @@ -185,6 +187,7 @@ def test_openai_chat_completion_stream_with_next_iteration(openai): assert generation.data[0].end_time is not None assert generation.data[0].start_time < generation.data[0].end_time assert generation.data[0].model_parameters == { + "service_tier": "default", "temperature": 0, "top_p": 1, "frequency_penalty": 0, @@ -391,6 +394,7 @@ def test_openai_chat_completion_with_seed(openai): ) assert generation.data[0].model_parameters == { + "service_tier": "default", "temperature": 0, "top_p": 1, "frequency_penalty": 0, @@ -659,6 +663,7 @@ async def test_async_chat(openai): assert generation.data[0].end_time is not None assert generation.data[0].start_time < generation.data[0].end_time assert generation.data[0].model_parameters == { + "service_tier": "default", "temperature": 1, "top_p": 1, "frequency_penalty": 0, @@ -703,6 +708,7 @@ async def test_async_chat_stream(openai): assert generation.data[0].end_time is not None assert generation.data[0].start_time < generation.data[0].end_time assert generation.data[0].model_parameters == { + "service_tier": "default", "temperature": 1, "top_p": 1, "frequency_penalty": 0, @@ -763,6 +769,7 @@ async def test_async_chat_stream_with_anext(openai): assert generation.data[0].end_time is not None assert generation.data[0].start_time < generation.data[0].end_time assert generation.data[0].model_parameters == { + "service_tier": "default", "temperature": 1, "top_p": 1, "frequency_penalty": 0, @@ -1054,6 +1061,7 @@ def test_structured_output_response_format_kwarg(openai): assert generation.data[0].end_time is not None assert generation.data[0].start_time < generation.data[0].end_time assert generation.data[0].model_parameters == { + "service_tier": "default", "temperature": 1, "top_p": 1, "frequency_penalty": 0, @@ -1160,6 +1168,7 @@ async def test_close_async_stream(openai): assert generation.data[0].end_time is not None assert generation.data[0].start_time < generation.data[0].end_time assert generation.data[0].model_parameters == { + "service_tier": "default", "temperature": 1, "top_p": 1, "frequency_penalty": 0,