core/agent: stream a repeated reasoning text only once

DeepSeek's Responses stream sends ``response.reasoning_text.delta`` fragments and then ``response.reasoning_text.done`` carrying the whole chain of thought those fragments spelled out. The loop folded a full delta into the round — it replaces the fragments, and that is what the assistant message hands back to the API — but streamed every delta verbatim, so the chain of thought was printed twice: once as it arrived and again as the ``.done`` text.

A full delta now streams only a tail the fragments did not cover, and nothing at all when it merely repeats them; a round whose fragments never arrived still streams the full text once. The message keeps the complete text either way.
This commit is contained in:
2026-10-07 00:26:38 +08:00
parent 4b5f67f5f3
commit bf8203ae37
2 changed files with 85 additions and 11 deletions
+25 -11
View File
@@ -260,16 +260,27 @@ async def _non_stream_round(
)
def _append_reasoning_delta(
def _take_reasoning_delta(
reasoning_parts: list[str],
delta: DeltaMessage,
) -> list[str]:
"""Accumulate a reasoning delta; a ``full`` (``.done``) event replaces fragments."""
if isinstance(delta.reasoning_content, str) and delta.reasoning_content:
if delta.reasoning_full is True:
return [delta.reasoning_content]
return [*reasoning_parts, delta.reasoning_content]
return reasoning_parts
) -> tuple[list[str], list[ReasoningDeltaEvent]]:
"""Fold a reasoning delta into the chain of thought; say what to stream.
Fragments append to *reasoning_parts*; a ``full`` delta replaces them, since
DeepSeek's ``response.reasoning_text.done`` carries the whole chain of thought
its fragments spelled out. The display already printed those fragments, so a
full delta streams only a tail they did not cover — emitting it verbatim
printed the CoT twice — and is dropped when it does not continue them at all.
A round that streamed nothing streams the full text once.
"""
content = delta.reasoning_content
if not isinstance(content, str) or not content:
return reasoning_parts, []
if delta.reasoning_full is not True:
return [*reasoning_parts, content], [ReasoningDeltaEvent(content=content)]
streamed = "".join(reasoning_parts)
fresh = content.removeprefix(streamed) if content.startswith(streamed) else ""
return [content], [ReasoningDeltaEvent(content=fresh)] if fresh else []
async def _stream_round(
@@ -312,9 +323,12 @@ async def _stream_round(
finish_reason = fr
saw_terminal = True
delta = choice.delta
if isinstance(delta.reasoning_content, str) and delta.reasoning_content:
reasoning_parts = _append_reasoning_delta(reasoning_parts, delta)
yield ReasoningDeltaEvent(content=delta.reasoning_content)
# A delta without reasoning folds to nothing, so every chunk can be taken
# here; the display sees only what it has not printed yet (see
# :func:`_take_reasoning_delta`).
reasoning_parts, reasoning_events = _take_reasoning_delta(reasoning_parts, delta)
for reasoning_event in reasoning_events:
yield reasoning_event
if isinstance(delta.content, str) and delta.content:
content_parts.append(delta.content)
yield TextDeltaEvent(content=delta.content)
@@ -9,6 +9,7 @@ from msgspec import UNSET
from plyngent.agent import (
AssistantMessageEvent,
ReasoningDeltaEvent,
TextDeltaEvent,
ToolCallEvent,
ToolRegistry,
@@ -258,6 +259,65 @@ async def test_dispatch_responses_stream_reasoning_no_duplicate() -> None:
assert reasoning == "streamed only"
async def _loop_reasoning_deltas(
events: list[ResponseStreamEvent],
messages: list[Any],
) -> list[str]:
"""Run one streamed round and collect the reasoning deltas the loop emitted."""
import msgspec
final = _completed_response(text="ok")
scripted = [*events, ResponseStreamEvent(type="response.completed", response=msgspec.to_builtins(final))]
client = ScriptedResponsesClient(stream_events=[scripted])
return [
event.content
async for event in run_chat_loop(cast("Any", client), messages, model="gpt-test", stream=True)
if isinstance(event, ReasoningDeltaEvent)
]
async def test_loop_responses_reasoning_done_is_not_printed_twice() -> None:
"""``response.reasoning_text.done`` repeats what its fragments spelled out.
Streaming it verbatim printed the chain of thought twice (the fragments, then
the whole text); the round still hands the full text back to the API.
"""
fragments = ["We", " need", " answer."]
events = [ResponseStreamEvent(type="response.reasoning_text.delta", delta=part) for part in fragments]
events.append(ResponseStreamEvent(type="response.reasoning_text.done", text="We need answer."))
messages: list[Any] = [UserChatMessage(content="hi")]
deltas = await _loop_reasoning_deltas(events, messages)
assert deltas == fragments
assert messages[-1].reasoning_content == "We need answer."
async def test_loop_responses_reasoning_done_alone_is_shown() -> None:
"""A round whose fragments never arrived still shows the ``.done`` text once."""
events = [ResponseStreamEvent(type="response.reasoning_text.done", text="chain of thought")]
messages: list[Any] = [UserChatMessage(content="hi")]
deltas = await _loop_reasoning_deltas(events, messages)
assert deltas == ["chain of thought"]
assert messages[-1].reasoning_content == "chain of thought"
async def test_loop_responses_reasoning_done_extends_fragments() -> None:
"""A ``.done`` text longer than what streamed shows only its unseen tail."""
events = [
ResponseStreamEvent(type="response.reasoning_text.delta", delta="We need"),
ResponseStreamEvent(type="response.reasoning_text.done", text="We need answer."),
]
messages: list[Any] = [UserChatMessage(content="hi")]
deltas = await _loop_reasoning_deltas(events, messages)
assert deltas == ["We need", " answer."]
assert messages[-1].reasoning_content == "We need answer."
def token_usage_present(chunk: object) -> bool:
usage = getattr(chunk, "usage", UNSET)
return usage is not UNSET and usage is not None