Emil_Shanaty/open-webui-ai4me
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: async DB calls, skip intermediate status writes, elif chain in event emitter (#22107)

Browse Source 
Three improvements to the socket event emitter hot path (when realtime chat save is enabled):

1. Wrap all synchronous Chats.* DB calls in asyncio.to_thread() to avoid blocking the event loop during streaming. With N concurrent users, sync DB calls serialize all writes and block socket event delivery.

2. Only persist final (done=True) status events to DB. Intermediate statuses (tool calling progress, web search progress, etc.) are ephemeral UI-only data already delivered via socket — writing every one to DB is unnecessary I/O.

3. Convert if/if/if chain to if/elif since event types are mutually exclusive, avoiding unnecessary string comparisons after a match.
This commit is contained in:
Classic298
2026-03-01 13:43:03 -05:00
committed by GitHub
parent 1db36b5eda
commit c436e0366c
1 changed files with 37 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/open_webui/socket/main.py
+37 -21
View File
@@ -796,15 +796,23 @@ def get_event_emitter(request_info, update_db=True):
and not request_info.get("chat_id", "").startswith("local:") and not request_info.get("chat_id", "").startswith("local:")
): ):
if "type" in event_data and event_data["type"] == "status": event_type = event_data.get("type")
Chats.add_message_status_to_chat_by_id_and_message_id(
request_info["chat_id"],
request_info["message_id"],
event_data.get("data", {}),
)
if "type" in event_data and event_data["type"] == "message": if event_type == "status":
message = Chats.get_message_by_id_and_message_id( # Only persist final (done) status to DB; intermediate
# statuses are ephemeral UI-only data sent via socket.
status_data = event_data.get("data", {})
if status_data.get("done", False):
await asyncio.to_thread(
Chats.add_message_status_to_chat_by_id_and_message_id,
request_info["chat_id"],
request_info["message_id"],
status_data,
)
elif event_type == "message":
message = await asyncio.to_thread(
Chats.get_message_by_id_and_message_id,
request_info["chat_id"], request_info["chat_id"],
request_info["message_id"], request_info["message_id"],
) )
@@ -813,7 +821,8 @@ def get_event_emitter(request_info, update_db=True):
content = message.get("content", "") content = message.get("content", "")
content += event_data.get("data", {}).get("content", "") content += event_data.get("data", {}).get("content", "")
Chats.upsert_message_to_chat_by_id_and_message_id( await asyncio.to_thread(
Chats.upsert_message_to_chat_by_id_and_message_id,
request_info["chat_id"], request_info["chat_id"],
request_info["message_id"], request_info["message_id"],
{ {
@@ -821,10 +830,11 @@ def get_event_emitter(request_info, update_db=True):
}, },
) )
if "type" in event_data and event_data["type"] == "replace": elif event_type == "replace":
content = event_data.get("data", {}).get("content", "") content = event_data.get("data", {}).get("content", "")
Chats.upsert_message_to_chat_by_id_and_message_id( await asyncio.to_thread(
Chats.upsert_message_to_chat_by_id_and_message_id,
request_info["chat_id"], request_info["chat_id"],
request_info["message_id"], request_info["message_id"],
{ {
@@ -832,8 +842,9 @@ def get_event_emitter(request_info, update_db=True):
}, },
) )
if "type" in event_data and event_data["type"] == "embeds": elif event_type == "embeds":
message = Chats.get_message_by_id_and_message_id( message = await asyncio.to_thread(
Chats.get_message_by_id_and_message_id,
request_info["chat_id"], request_info["chat_id"],
request_info["message_id"], request_info["message_id"],
) )
@@ -841,7 +852,8 @@ def get_event_emitter(request_info, update_db=True):
embeds = event_data.get("data", {}).get("embeds", []) embeds = event_data.get("data", {}).get("embeds", [])
embeds.extend(message.get("embeds", [])) embeds.extend(message.get("embeds", []))
Chats.upsert_message_to_chat_by_id_and_message_id( await asyncio.to_thread(
Chats.upsert_message_to_chat_by_id_and_message_id,
request_info["chat_id"], request_info["chat_id"],
request_info["message_id"], request_info["message_id"],
{ {
@@ -849,8 +861,9 @@ def get_event_emitter(request_info, update_db=True):
}, },
) )
if "type" in event_data and event_data["type"] == "files": elif event_type == "files":
message = Chats.get_message_by_id_and_message_id( message = await asyncio.to_thread(
Chats.get_message_by_id_and_message_id,
request_info["chat_id"], request_info["chat_id"],
request_info["message_id"], request_info["message_id"],
) )
@@ -858,7 +871,8 @@ def get_event_emitter(request_info, update_db=True):
files = event_data.get("data", {}).get("files", []) files = event_data.get("data", {}).get("files", [])
files.extend(message.get("files", [])) files.extend(message.get("files", []))
Chats.upsert_message_to_chat_by_id_and_message_id( await asyncio.to_thread(
Chats.upsert_message_to_chat_by_id_and_message_id,
request_info["chat_id"], request_info["chat_id"],
request_info["message_id"], request_info["message_id"],
{ {
@@ -866,10 +880,11 @@ def get_event_emitter(request_info, update_db=True):
}, },
) )
if event_data.get("type") in ["source", "citation"]: elif event_type in ("source", "citation"):
data = event_data.get("data", {}) data = event_data.get("data", {})
if data.get("type") == None: if data.get("type") is None:
message = Chats.get_message_by_id_and_message_id( message = await asyncio.to_thread(
Chats.get_message_by_id_and_message_id,
request_info["chat_id"], request_info["chat_id"],
request_info["message_id"], request_info["message_id"],
) )
@@ -877,7 +892,8 @@ def get_event_emitter(request_info, update_db=True):
sources = message.get("sources", []) sources = message.get("sources", [])
sources.append(data) sources.append(data)
Chats.upsert_message_to_chat_by_id_and_message_id( await asyncio.to_thread(
Chats.upsert_message_to_chat_by_id_and_message_id,
request_info["chat_id"], request_info["chat_id"],
request_info["message_id"], request_info["message_id"],
{ {