perf: make streamed content accumulation genuinely linear (#27359)

The streaming handler's content accumulator is a closure cell (declared nonlocal in stream_body_handler), and CPython's in-place string append optimization only applies to plain local variables (STORE_FAST), never to cell variables (STORE_DEREF). The content += value form introduced in #27231 therefore still allocates and copies the full accumulated string on every delta, exactly like the f-string it replaced; whether that copy is cheap or expensive is up to the allocator, and measurements swing accordingly (6 to 83 ms of pure copying for a 400 KB response on Python 3.12, against 0.3 ms for this patch).

Accumulate the deltas in a list instead and join once at the single read site (publish_chat_finished_event at stream end). List append is amortized O(1) with no dependence on reference counts, bytecode specialization or allocator behaviour, so accumulation is O(n) by construction. The non-str fallback keeps the previous f-string coercion semantics.

Verified end to end against a mock SSE upstream: a streamed chat with a think-tag block plus 40 content deltas produces output items, message text, reasoning text and usage identical to current dev, with no errors in the server log.
This commit is contained in:
Classic298 authored and GitHub committed 2026-07-23 18:17:41 -05:00
1 parent 9a49b271aa
commit 484fb61743
1 file changed
+8 -11
+8 -11
View File
@@ -3974,9 +3974,10 @@ async def streaming_chat_response_handler(response, ctx):
except Exception as e: except Exception as e:
pass pass
content = ( initial_content = (
message.get('content', '') if message else last_assistant_message if last_assistant_message else '' message.get('content', '') if message else last_assistant_message if last_assistant_message else ''
) )
content_parts = [initial_content] if initial_content else []
# Initialize output: use existing from message if continuing, else create new # Initialize output: use existing from message if continuing, else create new
existing_output = message.get('output') if message else None existing_output = message.get('output') if message else None
@@ -3984,14 +3985,14 @@ async def streaming_chat_response_handler(response, ctx):
output = existing_output output = existing_output
else: else:
# Only create an initial message item if there is content to initialize with # Only create an initial message item if there is content to initialize with
if content: if initial_content:
output = [ output = [
{ {
'type': 'message', 'type': 'message',
'id': output_id('msg'), 'id': output_id('msg'),
'status': 'in_progress', 'status': 'in_progress',
'role': 'assistant', 'role': 'assistant',
'content': [{'type': 'output_text', 'text': content}], 'content': [{'type': 'output_text', 'text': initial_content}],
} }
] ]
else: else:
@@ -4053,7 +4054,7 @@ async def streaming_chat_response_handler(response, ctx):
) )
async def stream_body_handler(response, form_data): async def stream_body_handler(response, form_data):
nonlocal content nonlocal content_parts
nonlocal usage nonlocal usage
nonlocal output nonlocal output
nonlocal prior_output nonlocal prior_output
@@ -4504,12 +4505,8 @@ async def streaming_chat_response_handler(response, ctx):
user, user,
) )
if isinstance(value, str): # closure-cell str += recopies per chunk; append + join once at read is O(n)
# In-place append — avoids copying the full content_parts.append(value if isinstance(value, str) else f'{value}')
# accumulated response on every chunk.
content += value
else:
content = f'{content}{value}'
# Check if we're inside a tag-based block # Check if we're inside a tag-based block
# (reasoning, code_interpreter, or solution). # (reasoning, code_interpreter, or solution).
@@ -5374,7 +5371,7 @@ async def streaming_chat_response_handler(response, ctx):
{'done': True}, {'done': True},
) )
await publish_chat_finished_event(request, user, metadata, title, content, output) await publish_chat_finished_event(request, user, metadata, title, ''.join(content_parts), output)
await event_emitter( await event_emitter(
{ {