diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index 7de42c00ede..1acffe95ccc 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -129,7 +129,9 @@ class LangfuseOtelLogger(OpenTelemetry): if not response_obj or not hasattr(response_obj, "get"): return - choices: Final = response_obj.get("choices", []) + observation_response: Final = response_obj.get("response") or response_obj + + choices: Final = observation_response.get("choices", []) if choices: first_choice: Final = choices[0] message: Final = first_choice.get("message", {}) @@ -144,7 +146,7 @@ class LangfuseOtelLogger(OpenTelemetry): except json.JSONDecodeError: arguments_obj = {} langfuse_tool_call = { - "id": response_obj.get("id", ""), + "id": observation_response.get("id", ""), "name": function.get("name", ""), "call_id": tool_call.get("id", ""), "type": "function_call", @@ -169,7 +171,7 @@ class LangfuseOtelLogger(OpenTelemetry): safe_dumps(output_data), ) - output: Final = response_obj.get("output", []) + output: Final = observation_response.get("output", []) if output: output_items_data: Final[list[dict]] = [] for item in output: diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 2e1e1a44594..9b70e5d995f 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -30,6 +30,10 @@ from litellm.litellm_core_utils.llm_response_utils.response_metadata import ( ) from litellm.litellm_core_utils.thread_pool_executor import executor from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig +from litellm.responses.sse_output_recovery import ( + record_output_item_chunk, + record_output_text_chunk, +) from litellm.responses.utils import ResponseAPILoggingUtils, ResponsesAPIRequestUtils from litellm.types.llms.openai import ( PART_UNION_TYPES, @@ -174,6 +178,8 @@ class BaseResponsesAPIStreamingIterator: self._completed_response_cache_hit: bool | None = None self._persist_completed_response_before_logging = True self._stream_created_time: float = time.time() + self._streamed_output_items: dict[int, dict[str, object]] = {} # mutable-ok: accumulated across SSE events + self._streamed_text_only_items: dict[int, dict[str, object]] = {} # mutable-ok: accumulated across SSE events # track request context for hooks self.litellm_metadata = litellm_metadata @@ -326,6 +332,24 @@ class BaseResponsesAPIStreamingIterator: openai_types.ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, openai_types.ResponsesAPIStreamEvents.RESPONSE_FAILED, ): + terminal_response_obj: Final[ResponsesAPIResponse | None] = getattr( + openai_responses_api_chunk, "response", None + ) + recovered_items: Final = { + **self._streamed_text_only_items, + **self._streamed_output_items, + } + if ( + terminal_response_obj is not None + and _chunk_type + in ( + openai_types.ResponsesAPIStreamEvents.RESPONSE_COMPLETED, + openai_types.ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, + ) + and not getattr(terminal_response_obj, "output", None) + and recovered_items + ): + terminal_response_obj.output = [item for _, item in sorted(recovered_items.items())] self.completed_response = openai_responses_api_chunk # Add cost to usage object if include_cost_in_streaming_usage is True if litellm.include_cost_in_streaming_usage and self.logging_obj is not None: @@ -561,6 +585,18 @@ class BaseResponsesAPIStreamingIterator: self._completed_response_cached = True + def _accumulate_streamed_output(self, chunk: ResponsesAPIStreamingResponse) -> None: + chunk_data: Final = chunk.model_dump() + event_type: Final = chunk_data.get("type") + if event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE: + record_output_item_chunk(chunk_data, self._streamed_output_items) + elif event_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE: + record_output_text_chunk( + chunk_data, + self._streamed_output_items, + self._streamed_text_only_items, + ) + async def _call_post_streaming_deployment_hook( self, chunk: ResponsesAPIStreamingResponse ) -> ResponsesAPIStreamingResponse: @@ -772,6 +808,7 @@ class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): result = await self._call_post_streaming_deployment_hook( chunk=result, ) + self._accumulate_streamed_output(result) self._yielded_first_chunk = True return result # If result is None, continue the loop to get the next chunk @@ -854,6 +891,7 @@ class SyncResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): async_function=self._call_post_streaming_deployment_hook, chunk=result, ) + self._accumulate_streamed_output(result) self._yielded_first_chunk = True return result # If result is None, continue the loop to get the next chunk diff --git a/tests/test_litellm/integrations/test_langfuse_otel.py b/tests/test_litellm/integrations/test_langfuse_otel.py index 28f138c7acd..f437b29cd76 100644 --- a/tests/test_litellm/integrations/test_langfuse_otel.py +++ b/tests/test_litellm/integrations/test_langfuse_otel.py @@ -6,7 +6,7 @@ import pytest from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger from litellm.integrations.opentelemetry import OpenTelemetryConfig -from litellm.types.llms.openai import ResponsesAPIResponse +from litellm.types.llms.openai import ResponseCompletedEvent, ResponsesAPIResponse class TestLangfuseOtelIntegration: @@ -826,6 +826,51 @@ class TestLangfuseOtelResponsesAPI: == "The weather in San Francisco is sunny, 20°C." ) + def test_streaming_responses_api_with_output(self): + from openai.types.responses import ResponseOutputMessage, ResponseOutputText + + from litellm.types.integrations.langfuse_otel import LangfuseSpanAttributes + + response_obj = ResponsesAPIResponse( + id="response-streaming", + created_at=1625247600, + output=[ + ResponseOutputMessage( + id="msg-streaming", + type="message", + role="assistant", + status="completed", + content=[ + ResponseOutputText( + annotations=[], + text="Streaming output is visible.", + type="output_text", + ) + ], + ) + ], + ) + completed_event = ResponseCompletedEvent( + type="response.completed", + response=response_obj, + ) + + with patch( + "litellm.integrations.arize._utils.safe_set_attribute" + ) as mock_safe_set_attribute: + LangfuseOtelLogger._set_langfuse_specific_attributes( + MagicMock(), {"call_type": "responses"}, completed_event + ) + + output_call = next( + call + for call in mock_safe_set_attribute.call_args_list + if call.args[1] == LangfuseSpanAttributes.OBSERVATION_OUTPUT.value + ) + assert json.loads(output_call.args[2]) == [ + {"role": "assistant", "content": "Streaming output is visible."} + ] + def test_responses_api_with_function_calls(self): """Test Langfuse OTEL logger with Responses API function_call output.""" from litellm.types.integrations.langfuse_otel import LangfuseSpanAttributes diff --git a/tests/test_litellm/responses/test_streaming_iterator.py b/tests/test_litellm/responses/test_streaming_iterator.py index 5b0f40fdf27..de19d93f5e4 100644 --- a/tests/test_litellm/responses/test_streaming_iterator.py +++ b/tests/test_litellm/responses/test_streaming_iterator.py @@ -6,13 +6,14 @@ completion_start_time = end_time.""" import json from datetime import datetime from typing import Optional -from unittest.mock import Mock +from unittest.mock import Mock, patch import httpx import pytest from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig +from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig from litellm.responses.streaming_iterator import ( ResponsesAPIStreamingIterator, SyncResponsesAPIStreamingIterator, @@ -235,3 +236,110 @@ def test_sync_transport_error_before_completed_event_raises(): with pytest.raises(httpx.ReadError): for _ in iterator: pass + + +def test_sync_stream_recovers_empty_completed_output(): + output_item = { + "type": "message", + "id": "msg_streaming", + "role": "assistant", + "status": "completed", + "content": [ + { + "type": "output_text", + "text": "Recovered streaming output", + "annotations": [], + } + ], + } + completed_response = { + "id": "resp_streaming", + "created_at": 1700000000, + "object": "response", + "status": "completed", + "model": "gpt-5.4", + "output": [], + } + response = httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=b"".join( + [ + _sse_event( + { + "type": "response.output_item.done", + "output_index": 0, + "item": output_item, + } + ), + _sse_event( + { + "type": "response.completed", + "response": completed_response, + } + ), + ] + ), + ) + iterator = SyncResponsesAPIStreamingIterator( + response=response, + model="gpt-5.4", + responses_api_provider_config=OpenAIResponsesAPIConfig(), + logging_obj=_logging_obj_stub(), + litellm_metadata={}, + custom_llm_provider="openai", + ) + + with patch.object(iterator, "_handle_logging_completed_response"): + events = list(iterator) + + completed_event = events[-1] + assert completed_event.response.output[0]["content"][0]["text"] == "Recovered streaming output" + + +@pytest.mark.asyncio +async def test_async_stream_recovers_output_text_done_when_completed_output_is_empty(): + response = httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=b"".join( + [ + _sse_event( + { + "type": "response.output_text.done", + "item_id": "msg_streaming", + "output_index": 0, + "content_index": 0, + "text": "Recovered text-only output", + } + ), + _sse_event( + { + "type": "response.completed", + "response": { + "id": "resp_streaming", + "created_at": 1700000000, + "object": "response", + "status": "completed", + "model": "gpt-5.4", + "output": [], + }, + } + ), + ] + ), + ) + iterator = ResponsesAPIStreamingIterator( + response=response, + model="gpt-5.4", + responses_api_provider_config=OpenAIResponsesAPIConfig(), + logging_obj=_logging_obj_stub(), + litellm_metadata={}, + custom_llm_provider="openai", + ) + + with patch.object(iterator, "_handle_logging_completed_response"): + events = [event async for event in iterator] + + completed_event = events[-1] + assert completed_event.response.output[0]["content"][0]["text"] == "Recovered text-only output"