From 28a191cd79682d468c75e89f08ee80b9562e6b5a Mon Sep 17 00:00:00 2001 From: worldmozara Date: Tue, 11 Aug 2026 20:08:52 +0800 Subject: [PATCH] core/agent: full reasoning_text.done replaces streamed fragments --- src/plyngent/agent/loop.py | 15 +++++++++- src/plyngent/agent/responses_dispatch.py | 29 ++++++++++++------- src/plyngent/agent/stream_chunks.py | 10 +++++-- .../lmproto/openai_compatible/model.py | 3 ++ tests/test_agent/test_responses_dispatch.py | 24 +++++++++++++-- 5 files changed, 66 insertions(+), 15 deletions(-) diff --git a/src/plyngent/agent/loop.py b/src/plyngent/agent/loop.py index 3a25391..4a1c71e 100644 --- a/src/plyngent/agent/loop.py +++ b/src/plyngent/agent/loop.py @@ -16,6 +16,7 @@ from plyngent.lmproto.openai_compatible.model import ( ChatCompletionChunk, ChatCompletionResponse, ChatCompletionsParam, + DeltaMessage, StreamOptions, StreamToolCallDelta, ToolChatMessage, @@ -254,6 +255,18 @@ async def _non_stream_round( ) +def _append_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 + + async def _stream_round( client: AnyLLMClient, param: ChatCompletionsParam, @@ -295,7 +308,7 @@ async def _stream_round( saw_terminal = True delta = choice.delta if isinstance(delta.reasoning_content, str) and delta.reasoning_content: - reasoning_parts.append(delta.reasoning_content) + reasoning_parts = _append_reasoning_delta(reasoning_parts, delta) yield ReasoningDeltaEvent(content=delta.reasoning_content) if isinstance(delta.content, str) and delta.content: content_parts.append(delta.content) diff --git a/src/plyngent/agent/responses_dispatch.py b/src/plyngent/agent/responses_dispatch.py index 68b43b1..36028ef 100644 --- a/src/plyngent/agent/responses_dispatch.py +++ b/src/plyngent/agent/responses_dispatch.py @@ -47,13 +47,15 @@ if TYPE_CHECKING: # Responses stream event types that carry reasoning text deltas. # OpenAI streams ``response.reasoning_summary_text.delta``; DeepSeek streams -# ``response.reasoning_text.delta`` and emits the full text once via -# ``response.reasoning_text.done``. +# ``response.reasoning_text.delta`` fragments and emits the FULL chain-of-thought +# once via ``response.reasoning_text.done``. Fragments append; a ``.done`` event +# carries the complete text and must replace the accumulated fragments (otherwise +# the CoT is concatenated twice). _REASONING_DELTA_TYPES = { "response.reasoning_summary_text.delta", "response.reasoning_text.delta", - "response.reasoning_text.done", } +_REASONING_FULL_TYPES = {"response.reasoning_text.done"} # Terminal events that carry the full Response object (status, tool calls, usage). # ``response.failed`` also carries a Response but is surfaced as an error. _FINAL_EVENT_TYPES = {"response.completed", "response.incomplete"} @@ -98,12 +100,19 @@ def _decode_final_event(event: ResponseStreamEvent) -> ResponseModel | None: return None -def _stream_reasoning_content(event: ResponseStreamEvent) -> str | None: - """Return reasoning text from a reasoning stream event (delta or full-text done).""" - if event.type not in _REASONING_DELTA_TYPES: - return None +def _stream_reasoning_content(event: ResponseStreamEvent) -> tuple[str | None, bool]: + """Return (reasoning_text, is_full) from a reasoning stream event. + + ``.delta`` events carry incremental fragments (``is_full=False``); ``.done`` + events carry the full chain-of-thought (``is_full=True``) and must replace + the accumulated fragments. + """ + if event.type not in _REASONING_DELTA_TYPES and event.type not in _REASONING_FULL_TYPES: + return None, False content = event.delta if isinstance(event.delta, str) else event.text - return content if isinstance(content, str) and content else None + if not isinstance(content, str) or not content: + return None, False + return content, event.type in _REASONING_FULL_TYPES def _terminal_chunks(final: ResponseModel, *, model: str) -> list[ChatCompletionChunk]: @@ -148,10 +157,10 @@ async def _stream_as_chat_chunks( if etype == "response.output_text.delta" and isinstance(event.delta, str) and event.delta: yield text_delta_chunk(model=model, content=event.delta) continue - reasoning = _stream_reasoning_content(event) + reasoning, reasoning_full = _stream_reasoning_content(event) if reasoning is not None: saw_reasoning = True - yield reasoning_delta_chunk(model=model, content=reasoning) + yield reasoning_delta_chunk(model=model, content=reasoning, full=reasoning_full) continue if final is None: diff --git a/src/plyngent/agent/stream_chunks.py b/src/plyngent/agent/stream_chunks.py index 7ac2ec8..cb55606 100644 --- a/src/plyngent/agent/stream_chunks.py +++ b/src/plyngent/agent/stream_chunks.py @@ -58,7 +58,13 @@ def text_delta_chunk(*, model: str, content: str, created: int = 0) -> ChatCompl ) -def reasoning_delta_chunk(*, model: str, content: str, created: int = 0) -> ChatCompletionChunk: +def reasoning_delta_chunk( + *, + model: str, + content: str, + created: int = 0, + full: bool = False, +) -> ChatCompletionChunk: return ChatCompletionChunk( id="bridge-stream", object="chat.completion.chunk", @@ -67,7 +73,7 @@ def reasoning_delta_chunk(*, model: str, content: str, created: int = 0) -> Chat choices=[ ChunkChoice( index=0, - delta=DeltaMessage(reasoning_content=content), + delta=DeltaMessage(reasoning_content=content, reasoning_full=full or UNSET), ) ], ) diff --git a/src/plyngent/lmproto/openai_compatible/model.py b/src/plyngent/lmproto/openai_compatible/model.py index b718478..568c22b 100644 --- a/src/plyngent/lmproto/openai_compatible/model.py +++ b/src/plyngent/lmproto/openai_compatible/model.py @@ -239,6 +239,9 @@ class DeltaMessage(Struct): # Providers often send content=null / reasoning_content=null on partial chunks. content: str | None | Unset = UNSET reasoning_content: str | None | Unset = UNSET + # True when reasoning_content is the FULL chain-of-thought (e.g. DeepSeek's + # ``response.reasoning_text.done``), replacing any streamed fragments. + reasoning_full: bool | Unset = UNSET tool_calls: list[StreamToolCallDelta] | Unset = UNSET diff --git a/tests/test_agent/test_responses_dispatch.py b/tests/test_agent/test_responses_dispatch.py index ddd9d35..d6808f6 100644 --- a/tests/test_agent/test_responses_dispatch.py +++ b/tests/test_agent/test_responses_dispatch.py @@ -186,7 +186,11 @@ async def test_dispatch_responses_stream_text_and_usage() -> None: async def _streamed_reasoning( events: list[ResponseStreamEvent], ) -> str: - """Run one Responses stream and join all streamed ``reasoning_content`` deltas.""" + """Run one Responses stream and join streamed ``reasoning_content`` deltas. + + Mirrors the agent loop: ``.done`` (``reasoning_full``) replaces accumulated + fragments instead of appending. + """ import msgspec final = _completed_response(text="ok") @@ -202,7 +206,10 @@ async def _streamed_reasoning( continue delta = chunk.choices[0].delta if isinstance(delta.reasoning_content, str) and delta.reasoning_content: - parts.append(delta.reasoning_content) + if delta.reasoning_full is True: + parts = [delta.reasoning_content] + else: + parts.append(delta.reasoning_content) return "".join(parts) @@ -224,6 +231,19 @@ async def test_dispatch_responses_stream_reasoning_deltas() -> None: assert reasoning == "step one step two" +async def test_dispatch_responses_stream_reasoning_full_replaces_fragments() -> None: + """A ``.done`` event with the full CoT must replace streamed fragments, not + append — otherwise the chain-of-thought is concatenated twice.""" + reasoning = await _streamed_reasoning( + [ + ResponseStreamEvent(type="response.reasoning_text.delta", delta="partial "), + ResponseStreamEvent(type="response.reasoning_text.delta", delta="thinking"), + ResponseStreamEvent(type="response.reasoning_text.done", delta="full chain of thought"), + ] + ) + assert reasoning == "full chain of thought" + + async def test_dispatch_responses_stream_terminal_only_reasoning() -> None: """DeepSeek sometimes delivers reasoning only on the terminal response.""" reasoning = await _streamed_reasoning([])