From fee3fe34b53080ee9680a3f2547c3b710c7acebf Mon Sep 17 00:00:00 2001 From: Classic298 <27028174+Classic298@users.noreply.github.com> Date: Wed, 3 Jun 2026 10:31:57 +0200 Subject: [PATCH] feat: run outlet filters for direct API callers (streaming + non-streaming) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: run outlet filters for direct API callers (streaming + non-streaming) outlet() filters previously only fired for Web-UI / chat invocations. Direct API calls to /api/chat/completions (and the Anthropic-compatible endpoint that routes through it) have no chat_id/message_id and no event channel, so outlet_filter_handler short-circuited and the streaming path took a pure passthrough branch that never accumulated the response. This wires outlet filters into the API path so they run end-to-end (e.g. for request/usage tracking), treating API callers like temp chats: the message list is built from form_data + the assistant response instead of the DB. Event emission and DB persistence stay skipped — event emitters cannot run on stateless API calls, which is an accepted limitation. - outlet_filter_handler: build the message list from form_data whenever there is no DB-backed chat (now covers API callers, not just temp/local/channel chats); only persist outlet edits for real persisted chats. - non_streaming_chat_response_handler: run outlet on the no-event_emitter (API) path; the response is returned unmodified. - streaming_chat_response_handler: in the passthrough branch, accumulate content + usage (Chat Completions and Responses API formats) and fire outlet once the stream ends. Gated on filter_functions to avoid per-chunk parsing for filter-less callers, and guarded so accumulation can never break the passthrough stream. --- backend/open_webui/utils/middleware.py | 116 +++++++++++++++++++++---- 1 file changed, 98 insertions(+), 18 deletions(-) diff --git a/backend/open_webui/utils/middleware.py b/backend/open_webui/utils/middleware.py index 63de31fbba..79bc44c4ce 100644 --- a/backend/open_webui/utils/middleware.py +++ b/backend/open_webui/utils/middleware.py @@ -3292,17 +3292,26 @@ 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: - return - is_temp_chat = chat_id.startswith('local:') or chat_id.startswith('channel:') + # Persisted chats read from the DB and get outlet edits written back; temp + # chats and direct API callers have no DB record, so build the message list + # from form_data instead (lets outlet fire for API callers, e.g. tracking). + is_db_chat = bool(chat_id) and not is_temp_chat and bool(message_id) + 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_db_chat: + messages_map = await Chats.get_messages_map_by_chat_id(chat_id) + if not messages_map: + return + + message_list = get_message_list(messages_map, message_id) + if not message_list: + return + else: + # Temp chat / API caller — build from form_data + assistant response. form_messages = ctx.get('form_data', {}).get('messages', []) assistant_message = ctx.get('assistant_message', {}) @@ -3323,14 +3332,6 @@ async def outlet_filter_handler(ctx): **assistant_message, } ) - else: - messages_map = await Chats.get_messages_map_by_chat_id(chat_id) - if not messages_map: - return - - message_list = get_message_list(messages_map, message_id) - if not message_list: - return model_id = model.get('id') if isinstance(model, dict) else model @@ -3383,10 +3384,9 @@ 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) + # Persist outlet edits + notify frontend (only persisted chats have a DB record) if outlet_result and outlet_result.get('messages'): - if not is_temp_chat and messages_map: + if is_db_chat and messages_map: for message in outlet_result['messages']: outlet_message_id = message.get('id') if outlet_message_id and outlet_message_id in messages_map: @@ -3562,6 +3562,23 @@ async def non_streaming_chat_response_handler(response, ctx): return response + # Direct API caller (no event channel): still run outlet filters (e.g. for + # tracking). Response returned as-is — no event emission, DB write, or edit-back. + try: + choices = response_data.get('choices', []) + if choices and choices[0].get('message', {}).get('content'): + content = response_data['choices'][0]['message']['content'] + usage = normalize_usage(response_data.get('usage', {}) or {}) + + ctx['assistant_message'] = { + 'content': content, + **({'output': response_data['output']} if response_data.get('output') else {}), + **({'usage': usage} if usage else {}), + } + await outlet_filter_handler(ctx) + except Exception as e: + log.debug(f'Error running outlet filters for API caller: {e}') + if isinstance(response, dict): response = merge_events_into_response(response_data, events) @@ -5220,11 +5237,57 @@ async def streaming_chat_response_handler(response, ctx): return await response_handler(response, events) else: - # Fallback to the original response + # Direct API caller (no event channel): stream through unchanged, but + # accumulate the response so outlet filters can fire at the end (e.g. for + # tracking). Gated on filter_functions to skip per-chunk parsing when none. + run_outlet = bool(filter_functions) + async def stream_wrapper(original_generator, events): def wrap_item(item): return f'data: {item}\n\n' + # Accumulated assistant response for the outlet filter. + outlet_output = [] + outlet_content = '' + outlet_usage = None + + def accumulate(raw): + """Best-effort parse of a chunk to capture final content + usage.""" + nonlocal outlet_output, outlet_content, outlet_usage + line = raw.decode('utf-8', 'replace') if isinstance(raw, bytes) else raw + if not isinstance(line, str): + return + for part in line.split('\n'): + part = part.strip() + if not part.startswith('data:'): + continue + part = part[len('data:') :].strip() + if not part or part == '[DONE]': + continue + try: + chunk = json.loads(part) + except Exception: + continue + if not isinstance(chunk, dict): + continue + + # Responses API events + if chunk.get('type', '').startswith('response.'): + outlet_output, meta = handle_responses_streaming_event(chunk, outlet_output) + if meta and meta.get('usage'): + outlet_usage = normalize_usage(meta['usage']) + continue + + # Chat Completions deltas + raw_usage = chunk.get('usage', {}) or {} + raw_usage.update(chunk.get('timings', {})) # llama.cpp + if raw_usage: + outlet_usage = normalize_usage(raw_usage) + for choice in chunk.get('choices', []): + piece = (choice.get('delta', {}) or {}).get('content') + if piece: + outlet_content += piece + for event in events: event, _ = await process_filter_functions( request=request, @@ -5238,6 +5301,12 @@ async def streaming_chat_response_handler(response, ctx): yield wrap_item(json.dumps(event)) async for data in original_generator: + if run_outlet: + try: + accumulate(data) + except Exception: + pass + data, _ = await process_filter_functions( request=request, filter_functions=filter_functions, @@ -5249,6 +5318,17 @@ async def streaming_chat_response_handler(response, ctx): if data: yield data + if run_outlet: + ctx['assistant_message'] = { + 'content': outlet_content or serialize_output(outlet_output), + **({'output': outlet_output} if outlet_output else {}), + **({'usage': outlet_usage} if outlet_usage else {}), + } + try: + await outlet_filter_handler(ctx) + except Exception as e: + log.debug(f'Error running outlet filters for API caller: {e}') + return StreamingResponse( stream_wrapper(response.body_iterator, events), headers=dict(response.headers),