From 3fbf2dc388fe1e240b5a97ec57a8e9dc7aa53505 Mon Sep 17 00:00:00 2001 From: worldmozara Date: Tue, 11 Aug 2026 15:27:01 +0800 Subject: [PATCH] core/agent: fix deepseek responses reasoning_content for tool calls --- src/plyngent/agent/responses_bridge.py | 45 +++++++++++------ src/plyngent/agent/responses_dispatch.py | 55 ++++++++++++++++---- tests/test_agent/test_responses_bridge.py | 53 +++++++++++++++++++ tests/test_agent/test_responses_dispatch.py | 56 ++++++++++++++++++++- 4 files changed, 183 insertions(+), 26 deletions(-) diff --git a/src/plyngent/agent/responses_bridge.py b/src/plyngent/agent/responses_bridge.py index 15c0022..2d3accb 100644 --- a/src/plyngent/agent/responses_bridge.py +++ b/src/plyngent/agent/responses_bridge.py @@ -171,7 +171,7 @@ def response_to_assistant_message(response: Response) -> AssistantChatMessage: ) for call in calls ] - reasoning = _reasoning_summary_text(response) + reasoning = reasoning_summary_text(response) return AssistantChatMessage( content=text or None, tool_calls=tool_calls, @@ -179,23 +179,40 @@ def response_to_assistant_message(response: Response) -> AssistantChatMessage: ) -def _reasoning_summary_text(response: Response) -> str: +def _reasoning_text_blocks(raw: object) -> list[str]: + """Collect text from a reasoning block list (``summary``/``content``).""" + parts: list[str] = [] + if not isinstance(raw, list): + return parts + for block_obj in cast("list[object]", raw): + if not isinstance(block_obj, dict): + continue + block_map = cast("dict[str, object]", block_obj) + if block_map.get("type") in {"summary_text", "output_text", "reasoning_content"}: + text = block_map.get("text") + if isinstance(text, str) and text: + parts.append(text) + return parts + + +def reasoning_summary_text(response: Response) -> str: + """Concatenate reasoning text from the response. + + OpenAI puts ``reasoning`` items in ``output`` with a ``summary`` block list. + DeepSeek (Responses convention) additionally returns the full chain-of-thought + in the top-level ``response.reasoning`` (``content``/``summary`` block lists); + prefer it when present so the two sources are not duplicated. + """ + reasoning = response.reasoning + if isinstance(reasoning, dict): + parts = _reasoning_text_blocks(reasoning.get("content")) + _reasoning_text_blocks(reasoning.get("summary")) + if parts: + return "".join(parts) parts: list[str] = [] for raw in response.output: if raw.get("type") != "reasoning": continue - summary = raw.get("summary") - if not isinstance(summary, list): - continue - summary_items = cast("list[object]", summary) - for block_obj in summary_items: - if not isinstance(block_obj, dict): - continue - block_map = cast("dict[str, object]", block_obj) - if block_map.get("type") in {"summary_text", "output_text"}: - text = block_map.get("text") - if isinstance(text, str) and text: - parts.append(text) + parts.extend(_reasoning_text_blocks(raw.get("summary"))) return "".join(parts) diff --git a/src/plyngent/agent/responses_dispatch.py b/src/plyngent/agent/responses_dispatch.py index 2c7e241..68b43b1 100644 --- a/src/plyngent/agent/responses_dispatch.py +++ b/src/plyngent/agent/responses_dispatch.py @@ -15,6 +15,7 @@ from msgspec import UNSET from plyngent.agent.responses_bridge import ( chat_param_to_responses_kwargs, + reasoning_summary_text, response_to_assistant_message, response_to_chat_completion, responses_status_to_finish_reason, @@ -45,7 +46,14 @@ if TYPE_CHECKING: ) # Responses stream event types that carry reasoning text deltas. -_REASONING_DELTA_TYPES = {"response.reasoning_summary_text.delta", "response.reasoning_text.delta"} +# OpenAI streams ``response.reasoning_summary_text.delta``; DeepSeek streams +# ``response.reasoning_text.delta`` and emits the full text once via +# ``response.reasoning_text.done``. +_REASONING_DELTA_TYPES = { + "response.reasoning_summary_text.delta", + "response.reasoning_text.delta", + "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"} @@ -90,6 +98,27 @@ 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 + content = event.delta if isinstance(event.delta, str) else event.text + return content if isinstance(content, str) and content else None + + +def _terminal_chunks(final: ResponseModel, *, model: str) -> list[ChatCompletionChunk]: + """Build the post-stream chunks (tool calls, finish_reason, usage).""" + assistant = response_to_assistant_message(final) + has_tools = assistant.tool_calls is not UNSET and bool(assistant.tool_calls) + finish = responses_status_to_finish_reason(final, has_tool_calls=has_tools) + chunks: list[ChatCompletionChunk] = list(tool_call_chunks_from_response(final, model=model)) + chunks.append(finish_reason_chunk(model=model, finish_reason=finish)) + usage = usage_chunk_from_response(final, model=model) + if usage is not None: + chunks.append(usage) + return chunks + + async def _stream_as_chat_chunks( client: OpenAIClient, create: ResponsesCreateParam, @@ -108,6 +137,7 @@ async def _stream_as_chat_chunks( """ stream = await client.responses(create, stream=True) final: ResponseModel | None = None + saw_reasoning = False async for event in stream: etype = event.type if etype in {"error", "response.failed"}: @@ -118,8 +148,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 - if etype in _REASONING_DELTA_TYPES and isinstance(event.delta, str) and event.delta: - yield reasoning_delta_chunk(model=model, content=event.delta) + reasoning = _stream_reasoning_content(event) + if reasoning is not None: + saw_reasoning = True + yield reasoning_delta_chunk(model=model, content=reasoning) continue if final is None: @@ -127,15 +159,16 @@ async def _stream_as_chat_chunks( # The loop will treat empty + no finish_reason as a glitch. return - assistant = response_to_assistant_message(final) - has_tools = assistant.tool_calls is not UNSET and bool(assistant.tool_calls) - finish = responses_status_to_finish_reason(final, has_tool_calls=has_tools) - for chunk in tool_call_chunks_from_response(final, model=model): + if not saw_reasoning: + # DeepSeek sometimes delivers reasoning only on the terminal response + # (top-level ``reasoning``); surface it so tool-call turns keep their + # reasoning_content when the assistant message goes back to the API. + terminal_reasoning = reasoning_summary_text(final) + if terminal_reasoning: + yield reasoning_delta_chunk(model=model, content=terminal_reasoning) + + for chunk in _terminal_chunks(final, model=model): yield chunk - yield finish_reason_chunk(model=model, finish_reason=finish) - usage = usage_chunk_from_response(final, model=model) - if usage is not None: - yield usage async def dispatch_responses( diff --git a/tests/test_agent/test_responses_bridge.py b/tests/test_agent/test_responses_bridge.py index 104e1ab..d37827c 100644 --- a/tests/test_agent/test_responses_bridge.py +++ b/tests/test_agent/test_responses_bridge.py @@ -1,8 +1,13 @@ +from __future__ import annotations + +from typing import Any + import msgspec from plyngent.agent.responses_bridge import ( chat_messages_to_responses_input, chat_param_to_responses_kwargs, + reasoning_summary_text, response_to_assistant_message, response_to_chat_completion, responses_status_to_finish_reason, @@ -145,6 +150,54 @@ def test_response_to_assistant_with_tools() -> None: assert completion.choices[0].finish_reason == "tool_calls" +def _reasoning_response( + *, + output_reasoning: bool = False, + top_level: dict[str, Any] | None = None, +) -> Response: + output: list[dict[str, Any]] = [] + if output_reasoning: + output.append( + { + "id": "reason_1", + "type": "reasoning", + "summary": [{"type": "summary_text", "text": "openai-style reasoning"}], + } + ) + return Response( + id="resp_r", + created_at=1, + model="deepseek-v4-flash", + output=output, + reasoning=top_level, + ) + + +def test_reasoning_summary_text_deepseek_top_level() -> None: + """DeepSeek Responses returns reasoning in the top-level ``reasoning`` field.""" + response = _reasoning_response(top_level={"content": [{"type": "reasoning_content", "text": "think step by step"}]}) + assert reasoning_summary_text(response) == "think step by step" + assistant = response_to_assistant_message(response) + assert assistant.reasoning_content == "think step by step" + + +def test_reasoning_summary_text_prefers_top_level() -> None: + response = _reasoning_response( + output_reasoning=True, + top_level={"summary": [{"type": "summary_text", "text": "top-level reasoning"}]}, + ) + assert reasoning_summary_text(response) == "top-level reasoning" + + +def test_reasoning_summary_text_falls_back_to_output_items() -> None: + response = _reasoning_response(output_reasoning=True) + assert reasoning_summary_text(response) == "openai-style reasoning" + + +def test_reasoning_summary_text_empty_when_absent() -> None: + assert reasoning_summary_text(_reasoning_response()) == "" + + def test_responses_status_to_finish_reason_incomplete() -> None: body = { "id": "resp_i", diff --git a/tests/test_agent/test_responses_dispatch.py b/tests/test_agent/test_responses_dispatch.py index 0cec7d2..ddd9d35 100644 --- a/tests/test_agent/test_responses_dispatch.py +++ b/tests/test_agent/test_responses_dispatch.py @@ -177,13 +177,67 @@ async def test_dispatch_responses_stream_text_and_usage() -> None: ) stream = await dispatch_responses(cast("Any", client), param, stream=True) chunks = [chunk async for chunk in cast("Any", stream)] - # text deltas + finish + usage texts = [c.choices[0].delta.content for c in chunks if c.choices and isinstance(c.choices[0].delta.content, str)] assert texts == ["stream", "ed"] assert any(c.choices and c.choices[0].finish_reason == "stop" for c in chunks) assert any(token_usage_present(c) for c in chunks) +async def _streamed_reasoning( + events: list[ResponseStreamEvent], +) -> str: + """Run one Responses stream and join all streamed ``reasoning_content`` deltas.""" + import msgspec + + final = _completed_response(text="ok") + final.reasoning = {"content": [{"type": "reasoning_content", "text": "terminal reasoning"}]} + events.append(ResponseStreamEvent(type="response.completed", response=msgspec.to_builtins(final))) + client = ScriptedResponsesClient(stream_events=[events]) + param = ChatCompletionsParam(model="gpt-test", messages=[UserChatMessage(content="hi")]) + stream = await dispatch_responses(cast("Any", client), param, stream=True) + chunks = [chunk async for chunk in cast("Any", stream)] + parts: list[str] = [] + for chunk in chunks: + if not chunk.choices: + continue + delta = chunk.choices[0].delta + if isinstance(delta.reasoning_content, str) and delta.reasoning_content: + parts.append(delta.reasoning_content) + return "".join(parts) + + +async def test_dispatch_responses_stream_reasoning_text_done() -> None: + """DeepSeek ``response.reasoning_text.done`` carries the full reasoning text.""" + reasoning = await _streamed_reasoning( + [ResponseStreamEvent(type="response.reasoning_text.done", delta="full chain of thought")] + ) + assert reasoning == "full chain of thought" + + +async def test_dispatch_responses_stream_reasoning_deltas() -> None: + reasoning = await _streamed_reasoning( + [ + ResponseStreamEvent(type="response.reasoning_text.delta", delta="step one "), + ResponseStreamEvent(type="response.reasoning_text.delta", delta="step two"), + ] + ) + assert reasoning == "step one step two" + + +async def test_dispatch_responses_stream_terminal_only_reasoning() -> None: + """DeepSeek sometimes delivers reasoning only on the terminal response.""" + reasoning = await _streamed_reasoning([]) + assert reasoning == "terminal reasoning" + + +async def test_dispatch_responses_stream_reasoning_no_duplicate() -> None: + """Streamed deltas win; the terminal reasoning is not re-emitted.""" + reasoning = await _streamed_reasoning( + [ResponseStreamEvent(type="response.reasoning_text.delta", delta="streamed only")] + ) + assert reasoning == "streamed only" + + def token_usage_present(chunk: object) -> bool: usage = getattr(chunk, "usage", UNSET) return usage is not UNSET and usage is not None