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/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, 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",