diff --git a/backend/open_webui/env.py b/backend/open_webui/env.py index f33e1f3278..a9f484c787 100644 --- a/backend/open_webui/env.py +++ b/backend/open_webui/env.py @@ -932,6 +932,7 @@ else: ENABLE_CHAT_RESPONSE_BASE64_IMAGE_URL_CONVERSION = ( os.getenv('ENABLE_CHAT_RESPONSE_BASE64_IMAGE_URL_CONVERSION', 'False').lower() == 'true' ) +ENABLE_API_OUTLET_FILTERS = os.getenv('ENABLE_API_OUTLET_FILTERS', 'True').lower() == 'true' # When enabled, uses a hardcoded extension-to-MIME dictionary as a last-resort # fallback when both mimetypes.guess_type() and file.meta.content_type fail to diff --git a/backend/open_webui/utils/middleware.py b/backend/open_webui/utils/middleware.py index 8f3b1d9afc..26764e1f7f 100644 --- a/backend/open_webui/utils/middleware.py +++ b/backend/open_webui/utils/middleware.py @@ -32,6 +32,7 @@ from open_webui.env import ( BYPASS_MODEL_ACCESS_CONTROL, CHAT_RESPONSE_MAX_TOOL_CALL_ITERATIONS, CHAT_RESPONSE_STREAM_DELTA_CHUNK_SIZE, + ENABLE_API_OUTLET_FILTERS, ENABLE_CHAT_RESPONSE_BASE64_IMAGE_URL_CONVERSION, ENABLE_QUERIES_CACHE, ENABLE_REALTIME_CHAT_SAVE, @@ -3082,6 +3083,44 @@ def build_response_object(response, response_data): return response +def update_assistant_message_from_stream(assistant_message, raw): + line = raw.decode('utf-8', 'replace') if isinstance(raw, bytes) else raw + if not isinstance(line, str): + return + + for raw_part in line.splitlines(): + part = raw_part.removeprefix('data:').strip() + if not part or part == '[DONE]': + continue + + try: + data = json.loads(part) + except Exception: + continue + + if not isinstance(data, dict): + continue + + if data.get('type', '').startswith('response.'): + output, meta = handle_responses_streaming_event(data, assistant_message.get('output', [])) + if output: + assistant_message['output'] = output + assistant_message['content'] = serialize_output(output) + if meta and meta.get('usage'): + assistant_message['usage'] = merge_usage(assistant_message.get('usage'), meta['usage']) + continue + + raw_usage = data.get('usage', {}) or {} + raw_usage.update(data.get('timings', {})) + if raw_usage: + assistant_message['usage'] = merge_usage(assistant_message.get('usage'), raw_usage) + + for choice in data.get('choices', []): + content = (choice.get('delta', {}) or {}).get('content') + if content: + assistant_message['content'] = assistant_message.get('content', '') + content + + async def get_system_oauth_token(request, user): """Get the system OAuth token for a user. @@ -3350,9 +3389,7 @@ async def outlet_filter_handler(ctx): Persists outlet-modified content to DB and emits a chat:outlet event so the frontend can sync its in-memory state. - For temp chats (local: prefix), messages are built from form_data - plus the assistant response message stored in ctx['assistant_message'], - since temp chats have no DB-persisted history. + For temp/API chats, messages are built from form_data plus ctx['assistant_message']. """ request = ctx['request'] user = ctx['user'] @@ -3364,17 +3401,17 @@ async def outlet_filter_handler(ctx): chat_id = metadata.get('chat_id', '') message_id = metadata.get('message_id') - if not chat_id or not message_id: + if not chat_id and not ctx.get('assistant_message'): return - is_temp_chat = chat_id.startswith('local:') or chat_id.startswith('channel:') + if not message_id: + message_id = output_id('msg') + is_temp_chat = chat_id.startswith('local:') or chat_id.startswith('channel:') try: messages_map = None - if is_temp_chat: - # Temp chats have no DB record — build message list from - # the in-memory form_data plus the assistant response. + if is_temp_chat or not chat_id: form_messages = ctx.get('form_data', {}).get('messages', []) assistant_message = ctx.get('assistant_message', {}) @@ -3386,7 +3423,6 @@ async def outlet_filter_handler(ctx): for m in form_messages ] - # Append the full assistant message (content, output, usage, etc.) if assistant_message: message_list.append( { @@ -3395,6 +3431,9 @@ async def outlet_filter_handler(ctx): **assistant_message, } ) + + if not message_list: + return else: messages_map = await Chats.get_messages_map_by_chat_id(chat_id) if not messages_map: @@ -3455,8 +3494,6 @@ async def outlet_filter_handler(ctx): extra_params=extra_params, ) - # Persist outlet-modified content and notify frontend - # (skip DB persistence for temp chats — they have no DB record) if outlet_result and outlet_result.get('messages'): if not is_temp_chat and messages_map: for message in outlet_result['messages']: @@ -3635,6 +3672,18 @@ async def non_streaming_chat_response_handler(response, ctx): return response + choices = response_data.get('choices', []) + output = response_data.get('output') + content = choices[0].get('message', {}).get('content') if choices else '' + if ENABLE_API_OUTLET_FILTERS and (content or output): + usage = normalize_usage(response_data.get('usage', {}) or {}) + ctx['assistant_message'] = { + 'content': content or serialize_output(output), + **({'output': output} if output else {}), + **({'usage': usage} if usage else {}), + } + await outlet_filter_handler(ctx) + if isinstance(response, dict): response = merge_events_into_response(response_data, events) @@ -5333,6 +5382,8 @@ async def streaming_chat_response_handler(response, ctx): def wrap_item(item): return f'data: {item}\n\n' + assistant_message = {} + for event in events: event, _ = await process_filter_functions( request=request, @@ -5355,8 +5406,14 @@ async def streaming_chat_response_handler(response, ctx): ) if data: + if ENABLE_API_OUTLET_FILTERS: + update_assistant_message_from_stream(assistant_message, data) yield data + if ENABLE_API_OUTLET_FILTERS and assistant_message: + ctx['assistant_message'] = assistant_message + await outlet_filter_handler(ctx) + return StreamingResponse( stream_wrapper(response.body_iterator, events), headers=dict(response.headers),