mirror of
https://github.com/open-webui/open-webui.git
synced 2026-09-07 08:27:05 +00:00
refac
This commit is contained in:
Timothy Jaeryang Baek
parent
ae01ef9c95
commit
77d2000eb7
2 changed files with 107 additions and 23 deletions
|
|
@ -1624,6 +1624,8 @@ async def chat_completion(
|
|||
|
||||
async def process_chat(request, form_data, user, metadata, model, tasks=None):
|
||||
try:
|
||||
# Capture the assistant before provider conversion strips structured output from temp chats.
|
||||
ctx = await build_chat_response_context(request, form_data, user, model, metadata, tasks, [])
|
||||
form_data, metadata, events = await process_chat_payload(request, form_data, user, metadata, model)
|
||||
|
||||
if await drain_approved_tool_calls(request, form_data, user, model, metadata):
|
||||
|
|
@ -1639,7 +1641,7 @@ async def chat_completion(
|
|||
if isinstance(response, JSONResponse) and response.status_code >= 400:
|
||||
raise Exception(get_response_error_detail(response))
|
||||
|
||||
ctx = await build_chat_response_context(request, form_data, user, model, metadata, tasks, events)
|
||||
ctx.update(form_data=form_data, metadata=metadata, events=events)
|
||||
|
||||
return await process_chat_response(response, ctx)
|
||||
except asyncio.CancelledError:
|
||||
|
|
|
|||
|
|
@ -3106,6 +3106,16 @@ async def get_event_emitter_and_caller(metadata):
|
|||
|
||||
async def build_chat_response_context(request, form_data, user, model, metadata, tasks, events):
|
||||
event_emitter, event_caller = await get_event_emitter_and_caller(metadata)
|
||||
assistant_message = None
|
||||
if metadata.get('assistant_message_id'):
|
||||
if is_saved_chat_id(metadata.get('chat_id')):
|
||||
assistant_message = await Chats.get_message_by_id_and_message_id(
|
||||
metadata['chat_id'], metadata['assistant_message_id']
|
||||
)
|
||||
else:
|
||||
messages = form_data.get('messages', [])
|
||||
if messages and messages[-1].get('role') == 'assistant':
|
||||
assistant_message = messages[-1]
|
||||
return {
|
||||
'request': request,
|
||||
'form_data': form_data,
|
||||
|
|
@ -3116,6 +3126,7 @@ async def build_chat_response_context(request, form_data, user, model, metadata,
|
|||
'events': events,
|
||||
'event_emitter': event_emitter,
|
||||
'event_caller': event_caller,
|
||||
'assistant_message': copy.deepcopy(assistant_message),
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -4014,6 +4025,7 @@ async def non_streaming_chat_response_handler(response, ctx):
|
|||
|
||||
chat_id = metadata.get('chat_id') or ''
|
||||
save_to_chat = is_saved_chat_id(chat_id)
|
||||
continuing = bool(metadata.get('assistant_message_id'))
|
||||
|
||||
if event_emitter:
|
||||
try:
|
||||
|
|
@ -4057,21 +4069,24 @@ async def non_streaming_chat_response_handler(response, ctx):
|
|||
response_output = response_data.get('output')
|
||||
content = choices[0].get('message', {}).get('content') if choices else ''
|
||||
|
||||
if choices and (content or response_output):
|
||||
if content or response_output:
|
||||
await event_emitter(
|
||||
{
|
||||
'type': 'chat:completion',
|
||||
'data': response_data,
|
||||
}
|
||||
)
|
||||
if (continuing and 'error' not in response_data) or (
|
||||
not continuing and choices and (content or response_output)
|
||||
):
|
||||
if content or response_output or continuing:
|
||||
if not continuing:
|
||||
await event_emitter(
|
||||
{
|
||||
'type': 'chat:completion',
|
||||
'data': response_data,
|
||||
}
|
||||
)
|
||||
|
||||
title = await Chats.get_chat_title_by_id(metadata['chat_id']) if save_to_chat else ''
|
||||
|
||||
# Use output from backend if provided (OR-compliant backends),
|
||||
# otherwise generate from response content
|
||||
if not response_output:
|
||||
choice_message = choices[0].get('message', {})
|
||||
choice_message = choices[0].get('message', {}) if choices else {}
|
||||
reasoning_content = choice_message.get('reasoning_content') or choice_message.get('reasoning')
|
||||
reasoning_details = get_reasoning_details(choice_message)
|
||||
response_output = []
|
||||
|
|
@ -4103,10 +4118,38 @@ async def non_streaming_chat_response_handler(response, ctx):
|
|||
}
|
||||
)
|
||||
|
||||
if continuing:
|
||||
message = ctx.get('assistant_message') or {}
|
||||
previous = message.get('output') or []
|
||||
if not previous and message.get('content'):
|
||||
previous = [
|
||||
{
|
||||
'type': 'message',
|
||||
'id': message.get('id') or output_id('msg'),
|
||||
'content': [{'type': 'output_text', 'text': message['content']}],
|
||||
}
|
||||
]
|
||||
response_output = list(response_output)
|
||||
if (
|
||||
previous
|
||||
and response_output
|
||||
and previous[-1].get('type') == response_output[0].get('type') == 'message'
|
||||
):
|
||||
response_output[0] = {
|
||||
**previous[-1],
|
||||
**response_output[0],
|
||||
'id': previous[-1].get('id'),
|
||||
'content': [*previous[-1].get('content', []), *response_output[0].get('content', [])],
|
||||
}
|
||||
previous = previous[:-1]
|
||||
response_output = previous + response_output
|
||||
content = get_output_text(response_output)
|
||||
|
||||
await event_emitter(
|
||||
{
|
||||
'type': 'chat:completion',
|
||||
'data': {
|
||||
**(response_data if continuing else {}),
|
||||
'done': True,
|
||||
'output': response_output,
|
||||
'title': title,
|
||||
|
|
@ -4198,6 +4241,7 @@ async def streaming_chat_response_handler(response, ctx):
|
|||
event_caller = ctx['event_caller']
|
||||
chat_id = metadata.get('chat_id') or ''
|
||||
save_to_chat = is_saved_chat_id(chat_id)
|
||||
continuing = bool(metadata.get('assistant_message_id'))
|
||||
|
||||
extra_params = {
|
||||
'__event_emitter__': event_emitter,
|
||||
|
|
@ -4508,9 +4552,13 @@ async def streaming_chat_response_handler(response, ctx):
|
|||
return output, end_flag
|
||||
|
||||
message = (
|
||||
await Chats.get_message_by_id_and_message_id(metadata['chat_id'], metadata['message_id'])
|
||||
if save_to_chat
|
||||
else None
|
||||
ctx.get('assistant_message')
|
||||
if continuing
|
||||
else (
|
||||
await Chats.get_message_by_id_and_message_id(metadata['chat_id'], metadata['message_id'])
|
||||
if save_to_chat
|
||||
else None
|
||||
)
|
||||
)
|
||||
|
||||
tool_calls = []
|
||||
|
|
@ -4538,7 +4586,7 @@ async def streaming_chat_response_handler(response, ctx):
|
|||
and prior_output[-1].get('status') == 'in_progress'
|
||||
):
|
||||
msg_parts = prior_output[-1].get('content', [])
|
||||
if not msg_parts or (len(msg_parts) == 1 and not msg_parts[0].get('text', '').strip()):
|
||||
if not msg_parts or (len(msg_parts) == 1 and not msg_parts[0].get('text', '')):
|
||||
prior_output.pop()
|
||||
output = []
|
||||
content_parts = []
|
||||
|
|
@ -4559,10 +4607,31 @@ async def streaming_chat_response_handler(response, ctx):
|
|||
else:
|
||||
output = []
|
||||
|
||||
if continuing and not prior_output:
|
||||
# Legacy text is also a prefix, never provider output_index 0.
|
||||
prior_output, output = output, []
|
||||
content_parts = []
|
||||
|
||||
usage = None
|
||||
last_response_id = None
|
||||
|
||||
def full_output():
|
||||
if (
|
||||
continuing
|
||||
and prior_output
|
||||
and output
|
||||
and prior_output[-1].get('type') == output[0].get('type') == 'message'
|
||||
):
|
||||
return [
|
||||
*prior_output[:-1],
|
||||
{
|
||||
**prior_output[-1],
|
||||
**output[0],
|
||||
'id': prior_output[-1].get('id'),
|
||||
'content': [*prior_output[-1].get('content', []), *output[0].get('content', [])],
|
||||
},
|
||||
*output[1:],
|
||||
]
|
||||
return prior_output + output if prior_output else output
|
||||
|
||||
def get_message_error_content(error):
|
||||
|
|
@ -4677,7 +4746,9 @@ async def streaming_chat_response_handler(response, ctx):
|
|||
response_stream_task_id,
|
||||
chat_id,
|
||||
metadata.get('message_id'),
|
||||
joined_content or get_output_text(current_stream_output),
|
||||
get_output_text(current_stream_output)
|
||||
if continuing
|
||||
else joined_content or get_output_text(current_stream_output),
|
||||
current_stream_output,
|
||||
)
|
||||
|
||||
|
|
@ -4710,8 +4781,10 @@ async def streaming_chat_response_handler(response, ctx):
|
|||
if delta_count >= threshold and last_delta_data:
|
||||
await event_emitter(
|
||||
{
|
||||
'type': 'response:completion',
|
||||
'data': last_delta_data,
|
||||
'type': 'chat:completion' if continuing else 'response:completion',
|
||||
'data': {'output': full_output(), 'type': last_delta_data.get('type')}
|
||||
if continuing
|
||||
else last_delta_data,
|
||||
}
|
||||
)
|
||||
await save_current_response_stream()
|
||||
|
|
@ -4760,8 +4833,10 @@ async def streaming_chat_response_handler(response, ctx):
|
|||
await flush_pending_delta_data()
|
||||
await event_emitter(
|
||||
{
|
||||
'type': 'response:completion',
|
||||
'data': get_response_completion_event_data(response_data),
|
||||
'type': 'chat:completion' if continuing else 'response:completion',
|
||||
'data': {'output': full_output()}
|
||||
if continuing
|
||||
else get_response_completion_event_data(response_data),
|
||||
}
|
||||
)
|
||||
await save_current_response_stream(stream_output)
|
||||
|
|
@ -5427,7 +5502,7 @@ async def streaming_chat_response_handler(response, ctx):
|
|||
|
||||
if output:
|
||||
# Clean up the last message item
|
||||
if output[-1].get('type') == 'message':
|
||||
if output[-1].get('type') == 'message' and not continuing:
|
||||
parts = output[-1].get('content', [])
|
||||
if parts and parts[-1].get('type') == 'output_text':
|
||||
parts[-1]['text'] = parts[-1]['text'].strip()
|
||||
|
|
@ -5988,7 +6063,7 @@ async def streaming_chat_response_handler(response, ctx):
|
|||
prior_output.pop()
|
||||
output = []
|
||||
await stream_body_handler(res, new_form_data)
|
||||
output[:0] = prior_output
|
||||
output = full_output()
|
||||
prior_output = []
|
||||
elif getattr(res, 'status_code', 200) >= 400:
|
||||
await emit_message_error(get_message_error_content(get_response_error_detail(res)))
|
||||
|
|
@ -6229,7 +6304,12 @@ async def streaming_chat_response_handler(response, ctx):
|
|||
|
||||
await clear_response_stream(request.app.state.redis, response_stream_task_id)
|
||||
await publish_chat_finished_event(
|
||||
request, user, metadata, title, ''.join(content_parts), current_output
|
||||
request,
|
||||
user,
|
||||
metadata,
|
||||
title,
|
||||
get_output_text(current_output) if continuing else ''.join(content_parts),
|
||||
current_output,
|
||||
)
|
||||
|
||||
await event_emitter(
|
||||
|
|
@ -6240,7 +6320,9 @@ async def streaming_chat_response_handler(response, ctx):
|
|||
)
|
||||
|
||||
ctx['assistant_message'] = {
|
||||
'content': ''.join(content_parts) or get_output_text(current_output),
|
||||
'content': get_output_text(current_output)
|
||||
if continuing
|
||||
else ''.join(content_parts) or get_output_text(current_output),
|
||||
'output': current_output,
|
||||
**({'usage': usage} if usage else {}),
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue