From bf8203ae3715906c3e717eef043faef2d46a5398 Mon Sep 17 00:00:00 2001 From: worldmozara Date: Wed, 7 Oct 2026 00:26:38 +0800 Subject: [PATCH] core/agent: stream a repeated reasoning text only once MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- src/plyngent/agent/loop.py | 36 +++++++++---- tests/test_agent/test_responses_dispatch.py | 60 +++++++++++++++++++++ 2 files changed, 85 insertions(+), 11 deletions(-) diff --git a/src/plyngent/agent/loop.py b/src/plyngent/agent/loop.py index 655b190..4bc72b5 100644 --- a/src/plyngent/agent/loop.py +++ b/src/plyngent/agent/loop.py @@ -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) diff --git a/tests/test_agent/test_responses_dispatch.py b/tests/test_agent/test_responses_dispatch.py index 03bdce1..17ea31a 100644 --- a/tests/test_agent/test_responses_dispatch.py +++ b/tests/test_agent/test_responses_dispatch.py @@ -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