perf: write task payloads to Redis as bytes (#28833)

Saving a streaming response serialized the payload with orjson, decoded it to
str, scanned it for the three Unicode line separators and let redis-py encode
it straight back to UTF-8: on an 8 MB non-ASCII chat that is 6.9 ms and ~22 MB
of transient buffers per write, synchronously on the event loop.

json_codec now exposes dumps_bytes, which returns the serialized payload as
UTF-8 bytes without the line-separator escaping, and the two Redis writes in
tasks.py use it. That escaping only protects line-framed protocols such as
SSE; every reader of these Redis values re-parses them before anything is
served, and the escaped and raw forms parse identically, so mixed versions
during a rolling deploy interoperate both ways. The same write drops to
0.9 ms and one 8 MB buffer (7.5x), with 31-66% saved on KB-sized writes.
With ENABLE_ORJSON off, dumps_bytes wraps stdlib json, behaviour unchanged.

The str path keeps the escaping but applies it with chained str.replace
instead of a translate table, cutting a separator-containing 8 MB payload
from 312 ms to 5.7 ms with byte-identical output.
This commit is contained in:
Classic298 authored and GitHub committed 2026-08-20 12:58:52 -07:00
1 parent a0e7d0e3a4
commit b0fdc00452
2 files changed
+22 -11

No files matched your search

+3 -3
View File
@@ -6,7 +6,7 @@ from uuid import uuid4
from redis.asyncio import Redis
from open_webui.env import REDIS_KEY_PREFIX
from open_webui.utils.json_codec import JSONCodec
from open_webui.utils.json_codec import JSONCodec, dumps_bytes
log = logging.getLogger(__name__)
@@ -77,7 +77,7 @@ async def redis_list_item_tasks(redis: Redis, item_id: str) -> list[str]:
async def redis_send_command(redis: Redis, command: dict):
command_json = JSONCodec.dumps(command)
command_json = dumps_bytes(command)
# RedisCluster doesn't expose publish() directly, but the
# PUBLISH command broadcasts across all cluster nodes server-side.
if hasattr(redis, 'nodes_manager'):
@@ -163,7 +163,7 @@ async def save_response_stream(
}
if redis:
await redis.hset(REDIS_RESPONSE_STREAMS_KEY, task_id, JSONCodec.dumps(data))
await redis.hset(REDIS_RESPONSE_STREAMS_KEY, task_id, dumps_bytes(data))
else:
response_streams[task_id] = data