diff --git a/litellm/llms/apodex/responses/transformation.py b/litellm/llms/apodex/responses/transformation.py index 065b7a475ec..c91ca316752 100644 --- a/litellm/llms/apodex/responses/transformation.py +++ b/litellm/llms/apodex/responses/transformation.py @@ -19,7 +19,8 @@ from __future__ import annotations from collections.abc import Mapping from time import time -from typing import TYPE_CHECKING, Final +from types import MappingProxyType +from typing import TYPE_CHECKING, Final, NamedTuple import httpx from pydantic import TypeAdapter, ValidationError @@ -45,6 +46,27 @@ _STATEFUL_PARAMS: Final = ("previous_response_id", "background") _CANCEL_RESPONSE_ADAPTER: Final = TypeAdapter(dict[str, object]) _BODY_FRAMING_HEADERS: Final = frozenset({"content-encoding", "content-length"}) +_SWARM_DELTA_EVENT: Final = "response.swarm.llm_delta" + + +class _SwarmChannel(NamedTuple): + """How one `swarm.data.channel` maps onto the OpenAI event that carries it.""" + + event_type: str + item_id_prefix: str + index_field: str + + +# A Deep Research run streams several agents. Only the reporter's `output_text` is +# the answer that lands in the final `response.completed` snapshot; the worker's +# channel-less deltas are an intermediate draft and must not be mistaken for it. +_SWARM_CHANNELS: Final = MappingProxyType( + { + "output_text": _SwarmChannel("response.output_text.delta", "msg", "content_index"), + "reasoning": _SwarmChannel("response.reasoning_summary_text.delta", "rs", "summary_index"), + } +) + class ApodexResponsesConfig(OpenAIResponsesAPIConfig): @property @@ -115,42 +137,49 @@ class ApodexResponsesConfig(OpenAIResponsesAPIConfig): parsed_chunk: dict, # mutable-ok: matches the base-class signature logging_obj: LiteLLMLoggingObj, ) -> ResponsesAPIStreamingResponse: - """Surface the Deep Research answer text as the OpenAI delta event callers expect. + """Surface a Deep Research run's text as the OpenAI delta events callers expect. - Observed live, not documented: a Deep Research stream carries its text in - `response.swarm.llm_delta` and never emits `response.output_text.delta`, so - without this the answer arrives only in the final `response.completed` - snapshot. `channel` splits the agent's reasoning from its answer; everything - else falls through to the base class as a GenericEvent. + Observed live, not documented: the stream carries all of its text in + `response.swarm.llm_delta` and never emits `response.output_text.delta` or + any reasoning event, so without this the answer arrives only in the final + `response.completed` snapshot and the reasoning is lost. Everything this + does not recognise, the remaining `response.swarm.*` lifecycle events + included, falls through to the base class as a GenericEvent. """ - swarm: Final = parsed_chunk.get("swarm") - swarm_data: Final = swarm.get("data") if isinstance(swarm, dict) else None - if ( - parsed_chunk.get("type") != "response.swarm.llm_delta" - or not isinstance(swarm_data, dict) - or swarm_data.get("channel") != "output_text" - or not isinstance(swarm_data.get("delta"), str) - ): - return super().transform_streaming_response( - model=model, - parsed_chunk=parsed_chunk, - logging_obj=logging_obj, - ) - - response_id: Final = str(parsed_chunk.get("response_id", "")) + mapped: Final = self._map_swarm_delta(parsed_chunk) return super().transform_streaming_response( model=model, - parsed_chunk={ # mutable-ok: JSON event payload - "type": "response.output_text.delta", - "item_id": f"msg_{response_id}", - "output_index": 0, - "content_index": 0, - "delta": swarm_data["delta"], - "sequence_number": parsed_chunk.get("sequence_number", 0), - }, + parsed_chunk=parsed_chunk if mapped is None else mapped, logging_obj=logging_obj, ) + @staticmethod + def _map_swarm_delta( + parsed_chunk: Mapping[str, object], + ) -> dict[str, object] | None: # mutable-ok: feeds the base class's `parsed_chunk: dict` + """The OpenAI event for this swarm delta, or None to pass the chunk through.""" + if parsed_chunk.get("type") != _SWARM_DELTA_EVENT: + return None + swarm: Final = parsed_chunk.get("swarm") + data: Final = swarm.get("data") if isinstance(swarm, Mapping) else None + if not isinstance(data, Mapping): + return None + + channel: Final = _SWARM_CHANNELS.get(data.get("channel")) + delta: Final = data.get("delta") + if channel is None or not isinstance(delta, str): + return None + + response_id: Final = str(parsed_chunk.get("response_id", "")) + return { # mutable-ok: JSON event payload + "type": channel.event_type, + "item_id": f"{channel.item_id_prefix}_{response_id}", + "output_index": 0, + channel.index_field: 0, + "delta": delta, + "sequence_number": parsed_chunk.get("sequence_number", 0), + } + def get_supported_openai_params(self, model: str) -> list: # mutable-ok: matches the base-class signature inherited: Final = super().get_supported_openai_params(model) if is_deep_research_model(model): diff --git a/tests/test_litellm/llms/apodex/test_apodex_responses_transformation.py b/tests/test_litellm/llms/apodex/test_apodex_responses_transformation.py index 789911126c3..e6bb831d2e5 100644 --- a/tests/test_litellm/llms/apodex/test_apodex_responses_transformation.py +++ b/tests/test_litellm/llms/apodex/test_apodex_responses_transformation.py @@ -246,7 +246,7 @@ class TestDeepResearchKeepsState: "response_id": "w_c4b77c96", "sequence_number": 12, "swarm": { - "agent_id": "stateful_react", + "agent_id": "reporter", "data": { "channel": "output_text", "delta": "Hello there, friend!", @@ -263,15 +263,46 @@ class TestDeepResearchKeepsState: assert event.item_id == "msg_w_c4b77c96" assert event.delta == "Hello there, friend!" assert event.sequence_number == 12 + assert event.content_index == 0 + + def test_reasoning_delta_becomes_a_reasoning_summary_delta(self): + """`response.reasoning_summary_text.delta` is what LiteLLM already translates + into an Anthropic `thinking_delta`, which is the route Deep Research takes on + /v1/messages. It also keeps a separate item id from the answer text.""" + config = _responses_config("apodex-1-1-deep-research") + event = config.transform_streaming_response( + model="apodex-1-1-deep-research", + parsed_chunk={ + "type": "response.swarm.llm_delta", + "response_id": "w_c4b77c96", + "sequence_number": 7, + "swarm": { + "agent_id": "stateful_react", + "data": {"channel": "reasoning", "delta": "The user wants", "delta_index": 0}, + }, + }, + logging_obj=None, + ) + + assert event.type == "response.reasoning_summary_text.delta" + assert event.item_id == "rs_w_c4b77c96" + assert event.delta == "The user wants" + assert event.summary_index == 0 + assert not hasattr(event, "content_index") @pytest.mark.parametrize( "channel", - ("reasoning", None), - ids=("reasoning-channel", "no-channel"), + (None, "tool_output"), + ids=("no-channel", "unknown-channel"), ) - def test_non_answer_deltas_are_not_claimed_as_output_text(self, channel): - """Most of the stream is the agent thinking; only `output_text` is the answer.""" - data = {"delta": "The"} if channel is None else {"channel": channel, "delta": "The"} + def test_intermediate_agent_deltas_are_not_claimed(self, channel): + """The worker agent streams a draft answer on a channel-less delta. + + Live capture: those four deltas spell "Hello, friend! How are you?" while the + reporter's `output_text` is the "Hello there, friend!" that lands in + response.completed. Claiming them would splice the draft into the answer. + """ + data = {"delta": "Hello,"} if channel is None else {"channel": channel, "delta": "Hello,"} config = _responses_config("apodex-1-1-deep-research") event = config.transform_streaming_response( model="apodex-1-1-deep-research", @@ -286,6 +317,33 @@ class TestDeepResearchKeepsState: assert event.type == "response.swarm.llm_delta" + @pytest.mark.parametrize( + "event_type", + ( + "response.swarm.run_started", + "response.swarm.run_finished", + "response.swarm.injection_window", + "response.swarm.llm_attempt_started", + "response.swarm.llm_attempt_finished", + ), + ) + def test_swarm_lifecycle_events_pass_through(self, event_type: str): + """LiteLLM cannot drop a chunk from the stream, so these stay as GenericEvent + rather than being silently swallowed; `run_finished` carries the final content.""" + config = _responses_config("apodex-1-1-deep-research") + event = config.transform_streaming_response( + model="apodex-1-1-deep-research", + parsed_chunk={ + "type": event_type, + "response_id": "w_c4b77c96", + "sequence_number": 4, + "swarm": {"agent_id": "reporter", "data": {"status": "success"}}, + }, + logging_obj=None, + ) + + assert event.type == event_type + def test_documented_events_pass_through_untouched(self): config = _responses_config("apodex-1-1-deep-research") event = config.transform_streaming_response(