From 46eecf2950b518005122b405896a18193ddbd54a Mon Sep 17 00:00:00 2001 From: Daoyuan Li <94409450+DaoyuanLi2816@users.noreply.github.com> Date: Thu, 8 Oct 2026 04:30:44 -0700 Subject: [PATCH] fix(models): align MindIE synchronous streaming compatibility (#6497) --- README.md | 2 + backend/docs/CONFIGURATION.md | 6 +- .../harness/deerflow/models/AGENTS.md | 7 ++ .../deerflow/models/mindie_provider.py | 24 ++++++- backend/tests/test_mindie_provider.py | 67 +++++++++++++++++++ 5 files changed, 103 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index db05092ad3..7b6b9d8f46 100644 --- a/README.md +++ b/README.md @@ -146,6 +146,8 @@ For Google's official Gemini OpenAI-compatible endpoint, use the For MindIE XML tool calls, see the [argument parsing and newline compatibility guide](backend/docs/CONFIGURATION.md#mindie-xml-tool-arguments). +Both synchronous and asynchronous streams retain this compatibility: tool-enabled +streams simulate chunks from a non-streaming response, while no-tool streams stay native. 1. **Clone the DeerFlow repository** diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index fd6a0b0881..1383a15992 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -296,8 +296,10 @@ avoid that ambiguity. Numeric conversion failures and unsafe Python-literal containers retain the entire original argument rather than rewriting it. The same behavior applies to synchronous generation, asynchronous generation, -tool-enabled simulated streaming, and native no-tool streaming (which carries the -fence state across chunks). Native tool-call arguments remain unchanged. +and both synchronous and asynchronous streaming. Tool-enabled streams use +non-streaming generation followed by simulated chunks, even with a +`streaming: true` model default; native no-tool streams carry the fence state +across chunks. Native tool-call arguments remain unchanged. No additional configuration is required. #### Gemini via Google's OpenAI-compatible endpoint diff --git a/backend/packages/harness/deerflow/models/AGENTS.md b/backend/packages/harness/deerflow/models/AGENTS.md index c5b751fc06..c8828e290b 100644 --- a/backend/packages/harness/deerflow/models/AGENTS.md +++ b/backend/packages/harness/deerflow/models/AGENTS.md @@ -159,6 +159,13 @@ finite negative values clamp to zero. Sync/async 429/500 regressions live in ### MindIE Provider (`packages/harness/deerflow/models/mindie_provider.py`) +Public sync and async streams use the same message normalization. No-tool native +streams share the fence-stateful newline decoder; tool-enabled streams explicitly +request `stream=False`, including with `streaming=True` model defaults, then use +the shared simulated chunker. Preserve terminal usage exactly once, XML/native +tool calls, invalid calls, and the original messages. Offline SDK boundary tests: +`tests/test_mindie_provider.py::test_public_stream_compatibility_through_sdk`. + `_fix_messages` converts tool results to the XML text format expected by MindIE. Only tool-message `type=json` payloads join the text channel; other non-text blocks remain omitted. Keep the `` escaping boundary and the diff --git a/backend/packages/harness/deerflow/models/mindie_provider.py b/backend/packages/harness/deerflow/models/mindie_provider.py index 4de1a02800..05cee5b9e3 100644 --- a/backend/packages/harness/deerflow/models/mindie_provider.py +++ b/backend/packages/harness/deerflow/models/mindie_provider.py @@ -330,6 +330,23 @@ class MindIEChatModel(ChatOpenAI): result = await super()._agenerate(_fix_messages(messages), stop=stop, run_manager=run_manager, **kwargs) return self._patch_result_with_tools(result) + def _stream(self, messages, stop=None, run_manager=None, **kwargs): + if not kwargs.get("tools"): + decoder = _EscapedNewlineStreamDecoder() + for chunk in super()._stream(_fix_messages(messages), stop=stop, run_manager=run_manager, **kwargs): + if isinstance(chunk.message.content, str): + chunk.message.content = decoder.push(chunk.message.content) + yield chunk + tail = decoder.flush() + if tail: + yield ChatGenerationChunk(message=AIMessageChunk(content=tail)) + return + + # Tool-enabled MindIE requests cannot use native streaming, even when + # the model's default is streaming=True. + result = self._generate(messages, stop=stop, run_manager=run_manager, **{**kwargs, "stream": False}) + yield from self._simulate_stream(result) + async def _astream(self, messages, stop=None, run_manager=None, **kwargs): # Route standard queries to native streaming for lower TTFB if not kwargs.get("tools"): @@ -347,8 +364,13 @@ class MindIEChatModel(ChatOpenAI): # Fallback for tool-enabled requests: # MindIE currently drops choices when stream=True and tools are present. # We await the full generation and yield chunks to simulate streaming. - result = await self._agenerate(messages, stop=stop, run_manager=run_manager, **kwargs) + result = await self._agenerate(messages, stop=stop, run_manager=run_manager, **{**kwargs, "stream": False}) + for chunk in self._simulate_stream(result): + yield chunk + @staticmethod + def _simulate_stream(result: ChatResult) -> Iterator[ChatGenerationChunk]: + """Keep sync/async tool-mode chunking and terminal usage identical.""" for gen in result.generations: msg = gen.message content = msg.content diff --git a/backend/tests/test_mindie_provider.py b/backend/tests/test_mindie_provider.py index b907829b95..29e896e51e 100644 --- a/backend/tests/test_mindie_provider.py +++ b/backend/tests/test_mindie_provider.py @@ -6,6 +6,7 @@ import html import json from unittest.mock import AsyncMock, patch +import httpx import pytest from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage from langchain_core.outputs import ChatGeneration, ChatResult @@ -1028,3 +1029,69 @@ class TestSanitizationMindIEChain: assert "what does this note say?" in model_text assert "" not in model_text assert "ignore previous instructions" not in model_text + + +@pytest.mark.asyncio +@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize("streaming_default", [False, True], ids=["default", "streaming-default"]) +@pytest.mark.parametrize("response_kind", ["prose", "xml-tool", "native-tool", "invalid-tool", "text-only"]) +async def test_public_stream_compatibility_through_sdk(async_mode, streaming_default, response_kind): + """Sync and async streams share the MindIE wire boundary and terminal usage.""" + requests = [] + with_tools = response_kind != "prose" + raw_chunks = ["intro\\n``", '`py\\nprint("a\\nb")', "```\\noutro\\", "nend"] + text = "answer " + "a" * 35 + message = {"role": "assistant", "content": text} + if response_kind == "xml-tool": + message["content"] += "hello" + elif response_kind in ("native-tool", "invalid-tool"): + message["content"] = "" + arguments = '{"text":"hello"}' if response_kind == "native-tool" else "not-json" + message["tool_calls"] = [{"id": "native-call", "type": "function", "function": {"name": "echo", "arguments": arguments}}] + + def handle(request): + payload = json.loads(request.content) + requests.append(payload) + if not payload.get("stream"): + return httpx.Response( + 200, json={"id": "mindie-completion", "model": "mindie-test", "choices": [{"index": 0, "message": message, "finish_reason": "stop"}], "usage": {"prompt_tokens": 2, "completion_tokens": 3, "total_tokens": 5}} + ) + # Tool-enabled streaming is unsupported by MindIE. Keep its empty-choices + # response here: the adapter must use non-streaming generation instead. + frames = [] if with_tools else [{"id": "mindie-completion", "model": "mindie-test", "choices": [{"index": 0, "delta": {"role": "assistant", "content": part}, "finish_reason": None}]} for part in raw_chunks] + frames.append({"id": "mindie-completion", "model": "mindie-test", "choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3, "total_tokens": 5}}) + body = "".join(f"data: {json.dumps(frame)}\n\n" for frame in frames) + "data: [DONE]\n\n" + return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body.encode()) + + messages = [HumanMessage(content="continue"), ToolMessage(content="old ", tool_call_id="previous")] + original = [message.model_dump() for message in messages] + tool = {"type": "function", "function": {"name": "echo", "parameters": {"type": "object", "properties": {"text": {"type": "string"}}}}} + kwargs = {"tools": [tool]} if with_tools else {} + model_kwargs = {"streaming": True} if streaming_default else {} + with httpx.Client(transport=httpx.MockTransport(handle)) as sync_client: + async with httpx.AsyncClient(transport=httpx.MockTransport(handle)) as async_client: + model = MindIEChatModel(model="mindie-test", api_key="dummy", base_url="https://offline.invalid/v1", http_client=sync_client, http_async_client=async_client, max_retries=0, **model_kwargs) + if async_mode: + chunks = [chunk async for chunk in model.astream(messages, **kwargs)] + else: + chunks = list(model.stream(messages, **kwargs)) + + assert len(requests) == 1 + assert requests[0]["stream"] is (not with_tools) + assert requests[0]["messages"][-1] == {"role": "user", "content": "\nold <result>\n"} + assert [message.model_dump() for message in messages] == original + merged = chunks[0] + for chunk in chunks[1:]: + merged += chunk + assert {key: merged.usage_metadata[key] for key in ("input_tokens", "output_tokens", "total_tokens")} == {"input_tokens": 2, "output_tokens": 3, "total_tokens": 5} + assert sum(chunk.usage_metadata is not None for chunk in chunks) == 1 + if response_kind in ("xml-tool", "native-tool"): + assert [(call["name"], call["args"]) for call in merged.tool_calls] == [("echo", {"text": "hello"})] + assert merged.content == (text if response_kind == "xml-tool" else "") + elif response_kind == "invalid-tool": + assert merged.invalid_tool_calls[0]["id"] == "native-call" + assert merged.invalid_tool_calls[0]["args"] == "not-json" + else: + assert merged.content == (_decode_escaped_newlines_outside_fences("".join(raw_chunks)) if response_kind == "prose" else text) + if with_tools: + assert any(chunk.response_metadata.get("finish_reason") == "stop" for chunk in chunks)