scriptoriumadmin/open-webui
1
0
Fork 0
mirror of https://github.com/open-webui/open-webui.git synced 2026-10-05 02:41:34 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

feat: run outlet filters for direct API callers (streaming + non-streaming)

Browse source 
* 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.
This commit is contained in:
Classic298 2026-06-03 10:31:57 +02:00 • committed by GitHub
parent 1a97751e37
commit fee3fe34b5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 98 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

116
backend/open_webui/utils/middleware.py
View file

@ -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),