scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-09 03:18:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 109

feat(apodex): stream Deep Research reasoning as reasoning deltas

Browse source 
A Deep Research run streams two agents. The worker emits its chain of
thought on the `reasoning` channel and a draft answer on a channel-less
delta; the reporter emits its own reasoning plus the single `output_text`
delta that matches the final response.completed snapshot.

Only `output_text` was mapped, so 176 of 181 deltas in a sample run
surfaced as GenericEvent and the reasoning was effectively lost. Map the
`reasoning` channel to response.reasoning_summary_text.delta, which
LiteLLM already translates into an Anthropic thinking_delta on the
/v1/messages route Deep Research takes, and give it its own item id.

The channel-less deltas stay unclaimed on purpose: splicing the worker's
draft into the answer would corrupt the text. The remaining
response.swarm.* lifecycle events keep passing through, since
transform_streaming_response has no way to drop a chunk and run_finished
carries the final content.
This commit is contained in:
zhanghanduo 2026-08-16 20:26:47 +08:00
parent 42dcfa12a2
commit ec3e293ac2
2 changed files with 123 additions and 36 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

89
litellm/llms/apodex/responses/transformation.py
View file

@ -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):

70
tests/test_litellm/llms/apodex/test_apodex_responses_transformation.py
View file

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