From 4216d89f12d49629166ff50cf74c5159f8d029ca Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Sun, 7 Dec 2025 00:12:52 +0100 Subject: [PATCH] fix: resolve streaming duplication and markdown formatting issues - Fix text duplication bug with proper delta calculation - Preserve markdown formatting with chunk-based delivery (50 chars) - Handle GeneratorExit errors from async context managers - Update Chat service streaming to preserve formatting - Ensure proper word-by-word streaming without duplicates --- src/chat/service.py | 11 +++--- src/responses/streaming.py | 73 +++++++++++++++++++++----------------- 2 files changed, 47 insertions(+), 37 deletions(-) diff --git a/src/chat/service.py b/src/chat/service.py index 21db8f7..f30a96b 100644 --- a/src/chat/service.py +++ b/src/chat/service.py @@ -214,9 +214,12 @@ async def create_chat_completion_stream( in_reasoning = False elif item.type == "message": - # Stream message content word by word + # Stream message content in chunks (preserves newlines, markdown, etc.) text = item.data["content"][0]["text"] - for word in text.split(): + chunk_size = 50 # characters per chunk + + for i in range(0, len(text), chunk_size): + chunk = text[i:i+chunk_size] yield ChatCompletionChunk( id=completion_id, object=constants.CHAT_COMPLETION_CHUNK_OBJECT, @@ -225,12 +228,12 @@ async def create_chat_completion_stream( choices=[ ChatCompletionChunkChoice( index=0, - delta=ChatCompletionChunkDelta(content=f"{word} "), + delta=ChatCompletionChunkDelta(content=chunk), finish_reason=None, ) ], ) - await asyncio.sleep(0.05) # Simulate typing + await asyncio.sleep(0.02) # Faster since chunks are larger # Final chunk with finish_reason yield ChatCompletionChunk( diff --git a/src/responses/streaming.py b/src/responses/streaming.py index 9f02356..c09a497 100644 --- a/src/responses/streaming.py +++ b/src/responses/streaming.py @@ -140,6 +140,7 @@ class StreamingCoordinator: import asyncio output_items = [] + last_message_text = "" # Track last streamed message text to compute deltas try: # Strip pipeline prefix if present (e.g., "pipeline.model" -> "model") @@ -189,45 +190,51 @@ class StreamingCoordinator: yield FunctionCallDone() elif item.type == "message": - # Stream output text with stop sequence and max tokens enforcement - text = item.data["content"][0]["text"] - words = text.split() + # Get current accumulated text from agent + current_text = item.data["content"][0]["text"] - # Track accumulated text and tokens for enforcement - accumulated_text = "" - output_tokens = 0 + # Only stream the NEW text (delta) since last update + if current_text.startswith(last_message_text): + # Extract only the new portion + delta_text = current_text[len(last_message_text):] - for word in words: - # Add word to accumulated text - word_with_space = f"{word} " - accumulated_text += word_with_space + if delta_text: + # Stream the delta text in chunks while preserving formatting + # (newlines, markdown, code blocks, etc.) + chunk_size = 50 # characters per chunk - # Check stop sequences - stop_found, text_before_stop = self._check_stop_sequence( - accumulated_text, - request.stop - ) + for i in range(0, len(delta_text), chunk_size): + chunk = delta_text[i:i+chunk_size] - if stop_found: - # Emit final text before stop sequence - remaining_text = text_before_stop[len(accumulated_text) - len(word_with_space):] - if remaining_text: - yield OutputTextDelta(delta=remaining_text) - yield OutputTextDone() - break + # Check stop sequences on full accumulated text + stop_found, text_before_stop = self._check_stop_sequence( + current_text, + request.stop + ) - # Check max tokens - output_tokens = self._count_tokens_approx(accumulated_text) - if self._check_max_tokens(output_tokens, request.max_output_tokens): - # Max tokens reached - stop streaming - yield OutputTextDone() - break + if stop_found: + # Only emit remaining delta before stop + remaining = text_before_stop[len(last_message_text):] + if remaining: + yield OutputTextDelta(delta=remaining) + yield OutputTextDone() + break - # Normal streaming - yield OutputTextDelta(delta=word_with_space) - await asyncio.sleep(0.05) # Simulate typing - else: - # Completed normally without stop/limit + # Check max tokens on full text + output_tokens = self._count_tokens_approx(current_text) + if self._check_max_tokens(output_tokens, request.max_output_tokens): + yield OutputTextDone() + break + + # Normal streaming of delta chunk (preserves all formatting) + yield OutputTextDelta(delta=chunk) + await asyncio.sleep(0.02) # Shorter delay since chunks are larger + + # Update tracking variable + last_message_text = current_text + + # If this is the final message (status=completed), ensure we send done + if item.data.get("status") == "completed": yield OutputTextDone() # Final response.done event with complete response