scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-05 08:07:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 136

Merge 399e9fb478 into 40423e6ec0

Browse source
This commit is contained in:
yunhungo 2026-08-26 23:15:29 +08:00 • committed by GitHub
commit 263f3b0a42
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 198 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
litellm/integrations/langfuse/langfuse_otel.py
View file

@ -134,7 +134,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", {})
@ -149,7 +151,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",
@ -174,7 +176,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:

38
litellm/responses/streaming_iterator.py
View file

@ -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,
@ -209,6 +213,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
@ -363,6 +369,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:
@ -600,6 +624,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:
@ -811,6 +847,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
@ -893,6 +930,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

47
tests/test_litellm/integrations/test_langfuse_otel.py
View file

@ -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:
@ -871,6 +871,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

110
tests/test_litellm/responses/test_streaming_iterator.py
View file

@ -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"