upstream-archive/open-webui
1
0
Fork 0
mirror of https://github.com/open-webui/open-webui.git synced 2026-07-23 09:10:55 -05:00
Code Issues Packages Projects Releases Wiki Activity

refac

Browse Source
This commit is contained in:
Timothy Jaeryang Baek
2026-06-29 01:01:27 -05:00
parent 0502494e9a
commit 7240517807
1 changed files with 47 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
backend/open_webui/utils/middleware.py
View File

@@ -4006,10 +4006,12 @@ async def streaming_chat_response_handler(response, ctx):
int(metadata.get('params', {}).get('stream_delta_chunk_size') or 1),
)
last_delta_data = None
last_delta_type = None
async def flush_pending_delta_data(threshold: int = 0):
nonlocal delta_count
nonlocal last_delta_data
nonlocal last_delta_type
if delta_count >= threshold and last_delta_data:
await event_emitter(
@@ -4020,6 +4022,22 @@ async def streaming_chat_response_handler(response, ctx):
)
delta_count = 0
last_delta_data = None
last_delta_type = None
async def queue_pending_delta_data(delta_data: dict, delta_type: str):
nonlocal delta_count
nonlocal last_delta_data
nonlocal last_delta_type
if last_delta_type and last_delta_type != delta_type:
await flush_pending_delta_data()
delta_count += 1
last_delta_data = delta_data
last_delta_type = delta_type
if delta_count >= delta_chunk_size:
await flush_pending_delta_data(delta_chunk_size)
async for line in response.body_iterator:
line = line.decode('utf-8', 'replace') if isinstance(line, bytes) else line
@@ -4068,11 +4086,16 @@ async def streaming_chat_response_handler(response, ctx):
)
# Check for Responses API events (type field starts with "response.")
elif data.get('type', '').startswith('response.'):
response_event_type = data.get('type', '')
response_event_is_delta = response_event_type.endswith('.delta')
output, response_metadata = handle_responses_streaming_event(data, output)
if not response_event_is_delta:
await flush_pending_delta_data()
# Emit citation sources from finalized output items
# (mirrors Chat Completions annotation handling at delta level)
if data.get('type') == 'response.output_item.done':
if response_event_type == 'response.output_item.done':
item = data.get('item', {})
if item.get('type') == 'message':
for part in item.get('content', []):
@@ -4131,12 +4154,21 @@ async def streaming_chat_response_handler(response, ctx):
processed_data.update(response_metadata)
processed_data.pop('done', None)
await event_emitter(
{
'type': 'chat:completion',
'data': processed_data,
}
)
if response_event_is_delta:
response_delta_type = response_event_type.split('.')[1]
await queue_pending_delta_data(
processed_data,
'tool_call'
if response_delta_type == 'function_call_arguments'
else 'content',
)
else:
await event_emitter(
{
'type': 'chat:completion',
'data': processed_data,
}
)
continue
else:
choices = data.get('choices', [])
@@ -4180,6 +4212,7 @@ async def streaming_chat_response_handler(response, ctx):
continue
delta = choices[0].get('delta', {})
delta_type = 'content'
# Handle delta annotations
annotations = delta.get('annotations')
@@ -4249,9 +4282,6 @@ async def streaming_chat_response_handler(response, ctx):
# Emit pending tool calls in real-time
if response_tool_calls:
# Flush any pending text first
await flush_pending_delta_data()
# Build pending function_call output items for display
pending_fc_items = []
for tc in response_tool_calls:
@@ -4268,14 +4298,10 @@ async def streaming_chat_response_handler(response, ctx):
}
)
await event_emitter(
{
'type': 'chat:completion',
'data': {
'content': serialize_output(full_output() + pending_fc_items),
},
}
)
data = {
'content': serialize_output(full_output() + pending_fc_items),
}
delta_type = 'tool_call'
image_urls = await get_image_urls(delta.get('images', []), request, metadata, user)
if image_urls:
@@ -4332,6 +4358,7 @@ async def streaming_chat_response_handler(response, ctx):
]
data = {'content': serialize_output(full_output())}
delta_type = 'content'
if value:
if (
@@ -4492,12 +4519,10 @@ async def streaming_chat_response_handler(response, ctx):
data = {
'content': serialize_output(full_output()),
}
delta_type = 'content'
if delta:
delta_count += 1
last_delta_data = data
if delta_count >= delta_chunk_size:
await flush_pending_delta_data(delta_chunk_size)
await queue_pending_delta_data(data, delta_type)
else:
await event_emitter(
{