From b8f9bdcd31c4327e3a52de1920b835be3ff139af Mon Sep 17 00:00:00 2001 From: Classic298 <27028174+Classic298@users.noreply.github.com> Date: Sat, 5 Sep 2026 17:00:54 +0200 Subject: [PATCH] perf: buffer the streamed response text and join it once When a chat has API outlet filters configured the streaming handler keeps its own copy of the assistant message, and every delta rebuilt that copy by concatenating the whole text again. The cost of a delta grows with the text already accumulated, so a long response gets slower the longer it runs, and a provider that sends large batched chunks makes it much worse. Deltas are now buffered in a list and joined once, when the stream has finished and the message is handed to the outlet filters. The join reproduces what the concatenation did without assuming the deltas are text: a stream filter can make a delta any type, and two lists used to merge into a list, so a buffer that is not all strings is folded with the same `+`. The buffered parts are tracked as they are created, so nothing walks the output tree at the end and a malformed part from a provider cannot raise where it did not before. Replaying streams through both revisions with 400-character deltas, 10000 deltas fall from 2317ms to 22ms and 50000 from 66.7s to 154ms; with 6-character deltas, 100000 deltas fall from 551ms to 148ms. Per-delta cost stops growing with the accumulated text. Holding the deltas until the stream ends costs about 49 bytes per delta, so half a megabyte on a ten thousand delta response and five megabytes on a hundred thousand. The other streaming paths still concatenate and are left alone here. --- backend/open_webui/utils/middleware.py | 35 ++++++++++++++++++++++++-- 1 file changed, 33 insertions(+), 2 deletions(-) diff --git a/backend/open_webui/utils/middleware.py b/backend/open_webui/utils/middleware.py index b9921e382d..f57b1a3f6a 100644 --- a/backend/open_webui/utils/middleware.py +++ b/backend/open_webui/utils/middleware.py @@ -3364,7 +3364,13 @@ def update_assistant_message_from_stream(assistant_message, raw): def append_output_text(item, text): parts = item.setdefault('content', []) if parts and parts[-1].get('type') == 'output_text': - parts[-1]['text'] += text + part = parts[-1] + buffer = part.get('_text_buffer') + if buffer is None: + buffer = [] + part['_text_buffer'] = buffer + assistant_message.setdefault('_buffered_parts', []).append(part) + buffer.append(text) else: parts.append({'type': 'output_text', 'text': text}) @@ -3382,6 +3388,7 @@ def update_assistant_message_from_stream(assistant_message, raw): continue if data.get('type', '').startswith('response.'): + flush_buffered_parts(assistant_message) output, meta = handle_responses_streaming_event(data, assistant_message.get('output', [])) if output: assistant_message['output'] = output @@ -3439,7 +3446,30 @@ def update_assistant_message_from_stream(assistant_message, raw): append_output_text(output[-1], content) - assistant_message['content'] = assistant_message.get('content', '') + content + assistant_message.setdefault('_content_buffer', []).append(content) + + +def join_buffered_text(text, buffer: list): + # A stream filter can make a delta any type, and the concatenation this replaces merged non-strings too. + if isinstance(text, str) and all(isinstance(chunk, str) for chunk in buffer): + return text + ''.join(buffer) + for chunk in buffer: + text += chunk + return text + + +def flush_buffered_parts(assistant_message: dict) -> None: + """Write the buffered text back into the output parts holding it.""" + for part in assistant_message.pop('_buffered_parts', []): + part['text'] = join_buffered_text(part['text'], part.pop('_text_buffer')) + + +def finalize_assistant_message(assistant_message: dict) -> None: + """Join the text buffered by update_assistant_message_from_stream.""" + flush_buffered_parts(assistant_message) + buffer = assistant_message.pop('_content_buffer', None) + if buffer: + assistant_message['content'] = join_buffered_text(assistant_message.get('content', ''), buffer) async def get_system_oauth_token(request, user): @@ -6108,6 +6138,7 @@ async def streaming_chat_response_handler(response, ctx): yield data if has_api_outlet_filters and assistant_message: + finalize_assistant_message(assistant_message) ctx['assistant_message'] = assistant_message await outlet_filter_handler(ctx)