mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
fix(models): keep escaped newlines inside fences in MindIE native stream (#6478)
The no-tool native stream applied _decode_escaped_newlines_outside_fences
to each streamed chunk independently, but a fenced code block almost never
opens and closes within one token-sized chunk, so the per-chunk fence regex
never saw a complete ``` pair and decoded every literal \n inside the fence.
Code like print("a\nb") reached the streamed answer and the persisted
message content with a real newline, while the non-streaming path and the
tool-mode simulated stream preserved it.
Decode the native stream with a fence-stateful decoder instead: it carries
whether the stream is inside a fence across chunks and holds back a
trailing partial ```/`\n` sequence, so the joined stream equals the one-shot
decode of the same reply. A fence left unclosed by a truncated reply keeps
its escapes, matching how an unterminated fence still renders as code.
Document native no-tool streaming in the MindIE decode contract.
This commit is contained in:
parent
4ef1f96487
commit
3772f7c7c3
@ -296,7 +296,8 @@ 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,
|
||||
and tool-enabled simulated streaming. Native tool-call arguments remain unchanged.
|
||||
tool-enabled simulated streaming, and native no-tool streaming (which carries the
|
||||
fence state across chunks). Native tool-call arguments remain unchanged.
|
||||
No additional configuration is required.
|
||||
|
||||
#### Gemini via Google's OpenAI-compatible endpoint
|
||||
|
||||
@ -217,6 +217,62 @@ def _decode_escaped_newlines_outside_fences(content: str) -> str:
|
||||
return "".join(parts)
|
||||
|
||||
|
||||
class _EscapedNewlineStreamDecoder:
|
||||
"""Decode literal `\\n` outside fenced code blocks across streamed chunks.
|
||||
|
||||
`_decode_escaped_newlines_outside_fences` needs the whole reply to pair
|
||||
``` delimiters, and a token-sized chunk almost never contains both, so
|
||||
decoding chunks independently would also decode escapes inside fences.
|
||||
This decoder carries the fence state between chunks and holds back a
|
||||
trailing partial ```/`\\n` sequence, so the joined stream equals the
|
||||
one-shot decode of the same reply. A fence the stream never closes keeps
|
||||
its escapes: an unterminated fence still renders as code.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._inside_fence = False
|
||||
self._pending = ""
|
||||
|
||||
def push(self, text: str) -> str:
|
||||
"""Return the text decodable so far, holding back an undecided tail."""
|
||||
data = self._pending + text
|
||||
self._pending = ""
|
||||
out = []
|
||||
while data:
|
||||
fence = data.find("```")
|
||||
if self._inside_fence:
|
||||
if fence == -1:
|
||||
# The closing ``` may be split across chunks: emit the
|
||||
# fenced code verbatim and keep a partial closer pending.
|
||||
hold = len(data) - len(data.rstrip("`"))
|
||||
out.append(data[: len(data) - hold])
|
||||
self._pending = data[len(data) - hold :]
|
||||
break
|
||||
# Fenced code up to and including its closing ``` stays verbatim.
|
||||
out.append(data[: fence + 3])
|
||||
data = data[fence + 3 :]
|
||||
self._inside_fence = False
|
||||
continue
|
||||
if fence == -1:
|
||||
# Outside a fence the tail may still grow into ``` or into an
|
||||
# escape split across the boundary; hold it back, decode the rest.
|
||||
hold = len(data) - len(data.rstrip("`\\"))
|
||||
out.append(_decode_escaped_newlines_outside_fences(data[: len(data) - hold]))
|
||||
self._pending = data[len(data) - hold :]
|
||||
break
|
||||
# Prose before the opening ``` decodes; the fence itself does not.
|
||||
out.append(_decode_escaped_newlines_outside_fences(data[:fence]))
|
||||
out.append(data[fence : fence + 3])
|
||||
data = data[fence + 3 :]
|
||||
self._inside_fence = True
|
||||
return "".join(out)
|
||||
|
||||
def flush(self) -> str:
|
||||
"""Return text held back for a partial ```/`\\n` at the end of the stream."""
|
||||
pending, self._pending = self._pending, ""
|
||||
return pending
|
||||
|
||||
|
||||
class MindIEChatModel(ChatOpenAI):
|
||||
"""Chat model adapter for MindIE engine.
|
||||
|
||||
@ -277,10 +333,15 @@ class MindIEChatModel(ChatOpenAI):
|
||||
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"):
|
||||
decoder = _EscapedNewlineStreamDecoder()
|
||||
async for chunk in super()._astream(_fix_messages(messages), stop=stop, run_manager=run_manager, **kwargs):
|
||||
if isinstance(chunk.message.content, str):
|
||||
chunk.message.content = _decode_escaped_newlines_outside_fences(chunk.message.content)
|
||||
chunk.message.content = decoder.push(chunk.message.content)
|
||||
yield chunk
|
||||
# A partial ```/`\n` at the very end must still reach the consumer.
|
||||
tail = decoder.flush()
|
||||
if tail:
|
||||
yield ChatGenerationChunk(message=AIMessageChunk(content=tail))
|
||||
return
|
||||
|
||||
# Fallback for tool-enabled requests:
|
||||
|
||||
@ -14,6 +14,7 @@ from langchain_core.outputs import ChatGeneration, ChatResult
|
||||
from deerflow.agents.middlewares.input_sanitization_middleware import InputSanitizationMiddleware
|
||||
from deerflow.models.mindie_provider import (
|
||||
MindIEChatModel,
|
||||
_decode_escaped_newlines_outside_fences,
|
||||
_fix_messages,
|
||||
_parse_xml_tool_call_to_dict,
|
||||
)
|
||||
@ -738,6 +739,90 @@ class TestAStream:
|
||||
|
||||
assert chunks[0].message.content == "a\nb"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_tools_stream_keeps_escaped_newlines_inside_fences(self):
|
||||
"""A fence spanning several chunks must still protect its literal `\\n`.
|
||||
|
||||
The one-shot decode needs a complete ```...``` pair to tell fenced code
|
||||
from prose; a token-sized chunk almost never contains both delimiters,
|
||||
so decoding each chunk independently rewrote code like print("a\\nb")
|
||||
into a real newline mid-stream. The joined stream must match the
|
||||
non-streaming decode of the same reply.
|
||||
"""
|
||||
from langchain_core.messages import AIMessageChunk
|
||||
from langchain_core.outputs import ChatGenerationChunk
|
||||
|
||||
raw_chunks = ['intro\\n```python\\nprint("a\\nb")', "\\n```\\noutro\\nend"]
|
||||
|
||||
async def fake_stream(*args, **kwargs):
|
||||
for text in raw_chunks:
|
||||
yield ChatGenerationChunk(message=AIMessageChunk(content=text))
|
||||
|
||||
with patch("deerflow.models.mindie_provider.ChatOpenAI._astream", side_effect=fake_stream), patch.object(MindIEChatModel, "__init__", return_value=None):
|
||||
model = MindIEChatModel.__new__(MindIEChatModel)
|
||||
chunks = await self._collect(model._astream([HumanMessage(content="write example code")]))
|
||||
|
||||
streamed = "".join(c.message.content for c in chunks)
|
||||
|
||||
# Code inside the fence keeps its literal backslash-n; prose outside decodes.
|
||||
assert 'print("a\\nb")' in streamed
|
||||
assert streamed.endswith("outro\nend")
|
||||
assert streamed == _decode_escaped_newlines_outside_fences("".join(raw_chunks))
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"raw_chunks",
|
||||
[
|
||||
# One chunk carries a complete fence pair: already decodable alone.
|
||||
["a\\n```py\\nx\\ny```\\nb"],
|
||||
# Fence opens and closes across chunk boundaries.
|
||||
['intro\\n```python\\nprint("a\\nb")', "\\n```\\noutro\\nend"],
|
||||
# Escape sequence split across the chunk boundary, outside any fence.
|
||||
["outside\\", "ncode"],
|
||||
# Fence delimiter itself split across the chunk boundary.
|
||||
["text\\n``", "`python\\nx\\ny", "\\n```\\nend"],
|
||||
# Two fences with prose between them, escaped prose still decodes.
|
||||
["```a\\n1```\\nmid\\n```b\\n2", "```\\ntail\\nend"],
|
||||
],
|
||||
ids=["single-chunk-fence", "fence-spans-chunks", "escape-split-across-chunks", "fence-delimiter-split", "two-fences"],
|
||||
)
|
||||
async def test_no_tools_stream_decode_matches_non_streaming(self, raw_chunks):
|
||||
"""The joined native stream equals the one-shot decode of the full reply."""
|
||||
from langchain_core.messages import AIMessageChunk
|
||||
from langchain_core.outputs import ChatGenerationChunk
|
||||
|
||||
async def fake_stream(*args, **kwargs):
|
||||
for text in raw_chunks:
|
||||
yield ChatGenerationChunk(message=AIMessageChunk(content=text))
|
||||
|
||||
with patch("deerflow.models.mindie_provider.ChatOpenAI._astream", side_effect=fake_stream), patch.object(MindIEChatModel, "__init__", return_value=None):
|
||||
model = MindIEChatModel.__new__(MindIEChatModel)
|
||||
chunks = await self._collect(model._astream([HumanMessage(content="q")]))
|
||||
|
||||
streamed = "".join(c.message.content for c in chunks)
|
||||
|
||||
assert streamed == _decode_escaped_newlines_outside_fences("".join(raw_chunks))
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_tools_stream_flushes_trailing_partial_fences(self):
|
||||
"""Text held back for a partial ```/`\\n` at stream end is not dropped."""
|
||||
from langchain_core.messages import AIMessageChunk
|
||||
from langchain_core.outputs import ChatGenerationChunk
|
||||
|
||||
raw_chunks = ["plain\\ntext", "``"]
|
||||
|
||||
async def fake_stream(*args, **kwargs):
|
||||
for text in raw_chunks:
|
||||
yield ChatGenerationChunk(message=AIMessageChunk(content=text))
|
||||
|
||||
with patch("deerflow.models.mindie_provider.ChatOpenAI._astream", side_effect=fake_stream), patch.object(MindIEChatModel, "__init__", return_value=None):
|
||||
model = MindIEChatModel.__new__(MindIEChatModel)
|
||||
chunks = await self._collect(model._astream([HumanMessage(content="q")]))
|
||||
|
||||
streamed = "".join(c.message.content for c in chunks)
|
||||
|
||||
assert streamed == _decode_escaped_newlines_outside_fences("".join(raw_chunks))
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_with_tools_fake_streams_text_in_chunks(self):
|
||||
with patch.object(MindIEChatModel, "_agenerate", new_callable=AsyncMock) as mock_ag, patch.object(MindIEChatModel, "__init__", return_value=None):
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user