This is an automated email from the ASF dual-hosted git repository.
shreemaan-abhishek pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/apisix.git
The following commit(s) were added to refs/heads/master by this push:
new a2d34cd62 feat(ai-cache): add streaming support with format tagging
(#13644)
a2d34cd62 is described below
commit a2d34cd62a0c8c8981acfa13f3a6967832569ebd
Author: Mohammad Izzraff Janius
<[email protected]>
AuthorDate: Fri Jul 3 19:30:02 2026 +0800
feat(ai-cache): add streaming support with format tagging (#13644)
---
apisix/plugins/ai-cache.lua | 223 +++---
apisix/plugins/ai-cache/key.lua | 11 +-
apisix/plugins/ai-cache/semantic.lua | 23 +-
apisix/plugins/ai-cache/stream.lua | 102 +++
apisix/plugins/ai-cache/vector-search/redis.lua | 11 +-
apisix/plugins/ai-providers/base.lua | 11 +
docs/en/latest/plugins/ai-cache.md | 8 +
docs/zh/latest/plugins/ai-cache.md | 8 +
t/lib/ai_cache_mock.lua | 15 +
t/plugin/ai-cache-semantic.t | 35 +-
t/plugin/ai-cache-streaming.t | 909 ++++++++++++++++++++++++
t/plugin/ai-cache.t | 4 +-
12 files changed, 1220 insertions(+), 140 deletions(-)
diff --git a/apisix/plugins/ai-cache.lua b/apisix/plugins/ai-cache.lua
index 895ccc7c5..5a09f0d85 100644
--- a/apisix/plugins/ai-cache.lua
+++ b/apisix/plugins/ai-cache.lua
@@ -21,6 +21,7 @@ local key_mod = require("apisix.plugins.ai-cache.key")
local binding = require("apisix.plugins.ai-protocols.binding")
local redis_util = require("apisix.utils.redis")
local semantic = require("apisix.plugins.ai-cache.semantic")
+local stream = require("apisix.plugins.ai-cache.stream")
local ngx = ngx
local ngx_null = ngx.null
@@ -81,6 +82,53 @@ local function release(conf, red)
end
+-- Run fn(red) on a pooled connection: released on success, closed when fn
+-- returns an error or throws. Returns fn's result, or (nil, err) on any
failure.
+local function with_redis(conf, fn)
+ local red, err = redis_util.new(conf)
+ if not red then
+ return nil, err
+ end
+ local ok, res, ferr = pcall(fn, red)
+ if not ok or ferr then
+ red:close()
+ return nil, not ok and res or ferr
+ end
+ release(conf, red)
+ return res
+end
+
+
+-- fail-open: a cache-backend or embedding failure must never break the
+-- request; log it and treat the lookup as a MISS.
+local function fail_open(ctx, what, err)
+ core.log.warn("ai-cache: ", what, ", fail-open as MISS: ", err)
+ ctx.ai_cache_status = "MISS"
+end
+
+
+-- The L1 stored value; encoded only here so the shape has one home.
+local function encode_entry(body, created_at, format)
+ return core.json.encode({ body = body, created_at = created_at, format =
format })
+end
+
+
+-- Best-effort L2 -> L1 backfill under this request's L1 key, carrying
+-- created_at and format so either layer replays the hit identically.
+local function backfill_l1(conf, ctx, red, hit)
+ local envelope = encode_entry(hit.body, hit.created_at, hit.format)
+ if not envelope then
+ core.log.warn("ai-cache: L1 backfill skipped: json.encode returned
nil")
+ return
+ end
+ local ok, err = red:set(ctx.ai_cache_key, envelope,
+ "EX", (conf.exact and conf.exact.ttl) or
DEFAULT_TTL)
+ if not ok then
+ core.log.warn("ai-cache: L1 backfill SET failed: ", err)
+ end
+end
+
+
local function serve_hit(conf, ctx, cached, similarity)
local status = "HIT"
ctx.ai_cache_status = status
@@ -93,7 +141,9 @@ local function serve_hit(conf, ctx, cached, similarity)
str_format("%.4f", similarity))
end
end
- core.response.set_header("Content-Type", "application/json")
+ core.response.set_header("Content-Type",
+ cached.format == stream.FORMAT_SSE and "text/event-stream"
+ or "application/json")
return core.response.exit(200, cached.body)
end
@@ -111,10 +161,11 @@ function _M.access(conf, ctx)
return
end
- -- Streaming responses are not cached in PR-1 (SSE replay is a later
- -- increment). ai-proxy (higher priority) has already classified the
- -- request, so bypass before doing any work.
- if ctx.var.request_type == "ai_stream" then
+ -- A stream on a non-SSE wire framing (bedrock's aws-eventstream) can never
+ -- be captured or replayed, so the lookup would be a guaranteed-miss redis
+ -- GET on every request: bypass before doing any work.
+ if ctx.var.request_type == "ai_stream"
+ and not stream.provider_capturable(ctx.picked_ai_instance) then
ctx.ai_cache_status = "BYPASS"
return
end
@@ -137,78 +188,67 @@ function _M.access(conf, ctx)
ctx.ai_cache_fingerprint = key_mod.fingerprint(ctx, body)
ctx.ai_cache_key = key_mod.build(conf, ctx, ctx.ai_cache_fingerprint)
- -- Remember which instance the fingerprint was computed for. ai-proxy-multi
- -- may fall back to a different instance in before_proxy; the log phase
uses
- -- this to avoid writing that fallback response under the original key.
+ -- which instance the fingerprint was computed for; log() checks it so a
+ -- fallback instance's response is never written under this key
ctx.ai_cache_picked_at_access = ctx.picked_ai_instance
- local red
- red, err = redis_util.new(conf)
- if not red then
- -- fail-open: never let a cache-backend outage break the request.
- core.log.warn("ai-cache: redis unavailable, fail-open as MISS: ", err)
- ctx.ai_cache_status = "MISS"
- return
- end
-
- local res
- res, err = red:get(ctx.ai_cache_key)
- if err then
- red:close()
- core.log.warn("ai-cache: redis get failed, fail-open as MISS: ", err)
- ctx.ai_cache_status = "MISS"
- return
- end
- if res ~= nil and res ~= ngx_null then
- local cached = core.json.decode(res)
- if cached and cached.body then
- release(conf, red)
- return serve_hit(conf, ctx, cached)
+ local cached
+ cached, err = with_redis(conf, function(red)
+ local res, gerr = red:get(ctx.ai_cache_key)
+ if gerr then
+ return nil, gerr
+ end
+ if res == nil or res == ngx_null then
+ return nil
+ end
+ local entry = core.json.decode(res)
+ if entry and entry.body then
+ return entry
end
core.log.warn("ai-cache: discarding malformed cache entry for ",
ctx.ai_cache_key)
+ return nil
+ end)
+ if err then
+ return fail_open(ctx, "L1 lookup failed", err)
+ end
+ if cached then
+ return serve_hit(conf, ctx, cached)
end
- -- L1 miss -> L2 semantic lookup. Release the L1 connection before
- -- embed_query()'s HTTP call so the pool isn't pinned across the embedding
- -- round-trip; re-acquire for the vector search. pcall keeps throws
fail-open.
+ -- L1 miss -> L2 semantic lookup, in its own connection scope so the pool
+ -- isn't pinned across embed_query()'s HTTP round-trip.
if has_layer(conf, "semantic") and conf.semantic then
- release(conf, red)
-
local ok, vec = pcall(semantic.embed_query, conf, ctx, body)
if not ok then
- core.log.warn("ai-cache: semantic embed error, fail-open as MISS:
", vec)
- vec = nil
+ fail_open(ctx, "semantic embed error", vec)
-- prevent log() from scheduling a write with partial/bad state
ctx.ai_cache_embedding = nil
+ return
end
if vec then
- local sred
- sred, err = redis_util.new(conf)
- if not sred then
- core.log.warn("ai-cache: redis unavailable for semantic
search, ",
- "fail-open as MISS: ", err)
- else
- local sok, hit = pcall(semantic.search, sred, conf, ctx, vec)
- if not sok then
- sred:close()
- core.log.warn("ai-cache: semantic search error, fail-open
as MISS: ", hit)
- ctx.ai_cache_embedding = nil
- else
- release(conf, sred)
- if hit then
- return serve_hit(conf, ctx,
- { body = hit.body, created_at = hit.created_at },
hit.similarity)
+ local hit
+ hit, err = with_redis(conf, function(red)
+ local h = semantic.search(red, conf, ctx, vec)
+ if h then
+ local bok, berr = pcall(backfill_l1, conf, ctx, red, h)
+ if not bok then
+ core.log.warn("ai-cache: L1 backfill error: ", berr)
end
end
+ return h
+ end)
+ if err then
+ fail_open(ctx, "semantic search failed", err)
+ ctx.ai_cache_embedding = nil
+ return
+ end
+ if hit then
+ return serve_hit(conf, ctx, hit, hit.similarity)
end
end
-
- ctx.ai_cache_status = "MISS"
- return
end
- release(conf, red)
ctx.ai_cache_status = "MISS"
end
@@ -222,7 +262,8 @@ end
function _M.body_filter(conf, ctx)
-- only a MISS gets written back; HIT exited in access, BYPASS opts out.
- if ctx.ai_cache_status ~= "MISS" or ctx.ai_cache_oversized then
+ if ctx.ai_cache_status ~= "MISS" or ctx.ai_cache_oversized
+ or not stream.capturable(ctx) then
return
end
local chunk = ngx.arg[1]
@@ -244,36 +285,30 @@ function _M.body_filter(conf, ctx)
end
--- The response-capturing phases (body_filter / log) run in contexts where
--- cosockets are disabled, so the Redis write is deferred to a 0-delay timer
--- (timers run in a light thread where cosockets are allowed).
--- l2 (optional) = { partition, embedding, dim, fingerprint, ttl } for L2
write.
-local function write_to_cache(premature, conf, cache_key, response_body, l2)
+-- body_filter/log cannot use cosockets, so the Redis write runs in a 0-delay
+-- timer. l2 (optional) = { partition, embedding, dim, fingerprint, ttl }.
+local function write_to_cache(premature, conf, cache_key, response_body, l2,
format)
if premature then
return
end
- local red, err = redis_util.new(conf)
- if not red then
- core.log.warn("ai-cache: redis unavailable on write: ", err)
- return
- end
- local envelope = core.json.encode({ body = response_body, created_at =
ngx.time() })
- local ttl = (conf.exact and conf.exact.ttl) or DEFAULT_TTL
- local ok
- ok, err = red:set(cache_key, envelope, "EX", ttl)
- if not ok then
- red:close()
- core.log.warn("ai-cache: redis set failed: ", err)
- return
- end
- if l2 then
- l2.created_at = ngx.time()
- local wok, werr = pcall(semantic.write, red, conf, l2, response_body)
- if not wok then
- core.log.warn("ai-cache: semantic write error: ", werr)
+ local now = ngx.time()
+ local envelope = encode_entry(response_body, now, format)
+ local _, err = with_redis(conf, function(red)
+ local ok, serr = red:set(cache_key, envelope, "EX",
+ (conf.exact and conf.exact.ttl) or
DEFAULT_TTL)
+ if not ok then
+ return nil, serr
+ end
+ if l2 then
+ l2.created_at = now
+ l2.format = format
+ semantic.write(red, conf, l2, response_body)
end
+ return true
+ end)
+ if err then
+ core.log.warn("ai-cache: cache write failed: ", err)
end
- release(conf, red)
end
@@ -281,27 +316,32 @@ function _M.log(conf, ctx)
if ctx.ai_cache_status ~= "MISS" or not ctx.ai_cache_fingerprint then
return
end
- -- ai-proxy-multi may reassign the picked instance on fallback/retry during
- -- before_proxy. The frozen fingerprint identifies the ORIGINAL instance,
so a
- -- response actually produced by a different (fallback) instance must not
be
- -- written under it -- that would replay the wrong instance's response on a
- -- later hit.
+ -- the fingerprint identifies the instance picked at access time; a
+ -- fallback/retry response from another instance must not be cached under
it
if ctx.picked_ai_instance ~= ctx.ai_cache_picked_at_access then
return
end
if ngx.status ~= 200 then
return
end
+ if ctx.ai_stream_aborted then
+ return
+ end
local buf = ctx.ai_cache_buf
if not buf or buf.bytes == 0 then
return
end
local response_body = concat(buf, "", 1, buf.n)
+ local format = stream.capture_format(ctx, response_body)
+ if not format then
+ return
+ end
+
local cache_key = key_mod.build(conf, ctx, ctx.ai_cache_fingerprint)
- -- Build the L2 doc from ctx fields stashed by semantic.embed_query(); the
- -- embedding is only set on a successful embed, so a nil check guards the
write.
+ -- L2 doc from ctx fields stashed by semantic.embed_query(); the embedding
+ -- is only set on a successful embed.
local l2
if has_layer(conf, "semantic") and ctx.ai_cache_embedding then
l2 = {
@@ -313,7 +353,8 @@ function _M.log(conf, ctx)
}
end
- local ok, err = ngx.timer.at(0, write_to_cache, conf, cache_key,
response_body, l2)
+ local ok, err = ngx.timer.at(0, write_to_cache, conf, cache_key,
+ response_body, l2, format)
if not ok then
core.log.warn("ai-cache: failed to schedule cache write: ", err)
end
diff --git a/apisix/plugins/ai-cache/key.lua b/apisix/plugins/ai-cache/key.lua
index f28cd63bb..9c6542d28 100644
--- a/apisix/plugins/ai-cache/key.lua
+++ b/apisix/plugins/ai-cache/key.lua
@@ -66,6 +66,8 @@ local function build_repr(ctx, body, messages)
params[k] = v
end
end
+ local proto = ctx.ai_client_protocol and
protocols.get(ctx.ai_client_protocol)
+ params.stream = (proto and proto.is_streaming(body)) == true
return {
client = {
@@ -111,15 +113,6 @@ function _M.fingerprint(ctx, body)
end
--- Returns the SHA-256 hex digest of the effective context with message text
--- removed. Queries that differ only in phrasing (same model/params/instance)
--- share one fingerprint, enabling semantic deduplication without storing the
--- raw prompt.
-function _M.context_fingerprint(ctx, body)
- return hex_digest(core.json.canonical_encode(build_repr(ctx, body, nil)))
-end
-
-
-- Percent-encode "%", ":" and "=" (in that order) in scope values so a
request-controlled
-- include_vars value can't shift "name=value:" boundaries to forge another
scope.
local function esc(v)
diff --git a/apisix/plugins/ai-cache/semantic.lua
b/apisix/plugins/ai-cache/semantic.lua
index dfaf4d0d7..6cb580283 100644
--- a/apisix/plugins/ai-cache/semantic.lua
+++ b/apisix/plugins/ai-cache/semantic.lua
@@ -282,7 +282,8 @@ end
-- Phase 2 of the L2 lookup: vector search over a caller-owned connection
-- acquired AFTER embed_query() (so the pool isn't pinned across embedding).
--- Returns a hit {body, created_at, similarity} on a >=threshold match, else
nil.
+-- Returns a hit {body, created_at, format, similarity} on a >=threshold match,
+-- else nil. The L1 backfill of a hit is the caller's job (ai-cache.lua owns
L1).
function _M.search(red, conf, ctx, vec)
local sem = conf.semantic
local target = redis_target(conf)
@@ -309,26 +310,13 @@ function _M.search(red, conf, ctx, vec)
return nil
end
- -- L2 -> L1 backfill, carrying the L2 entry's original created_at so Age is
- -- consistent whether the next hit is served from L1 or L2. A real
semantic
- -- hit must be served regardless — only the backfill SET is skipped on
error.
- local envelope = core.json.encode({ body = hit.response, created_at =
hit.created_at })
- if not envelope then
- core.log.warn("ai-cache: L1 backfill skipped: json.encode returned
nil")
- else
- local exact_ttl = (conf.exact and conf.exact.ttl) or 3600
- local bok, berr = red:set(ctx.ai_cache_key, envelope, "EX", exact_ttl)
- if not bok then
- core.log.warn("ai-cache: L1 backfill SET failed: ", berr)
- end
- end
-
- return { body = hit.response, created_at = hit.created_at, similarity =
similarity }
+ return { body = hit.response, created_at = hit.created_at,
+ format = hit.format, similarity = similarity }
end
-- Called from the write-back timer (after the L1 SET) with a still-open `red`.
--- l2 = { partition, embedding, dim, fingerprint, ttl, created_at }
+-- l2 = { partition, embedding, dim, fingerprint, ttl, created_at, format }
function _M.write(red, conf, l2, response_body)
if not l2 or not l2.embedding then
return
@@ -347,6 +335,7 @@ function _M.write(red, conf, l2, response_body)
embedding = vs.pack_float32(l2.embedding),
response = response_body,
created_at = l2.created_at,
+ format = l2.format,
}, l2.ttl)
if not ok then
core.log.warn("ai-cache: L2 upsert failed: ", err)
diff --git a/apisix/plugins/ai-cache/stream.lua
b/apisix/plugins/ai-cache/stream.lua
new file mode 100644
index 000000000..a91fdcb3c
--- /dev/null
+++ b/apisix/plugins/ai-cache/stream.lua
@@ -0,0 +1,102 @@
+--
+-- Licensed to the Apache Software Foundation (ASF) under one or more
+-- contributor license agreements. See the NOTICE file distributed with
+-- this work for additional information regarding copyright ownership.
+-- The ASF licenses this file to You under the Apache License, Version 2.0
+-- (the "License"); you may not use this file except in compliance with
+-- the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+--
+--
+-- Streaming (SSE) capture + replay policy for ai-cache: body_filter buffers
+-- the raw client-wire frames, log() stores them once complete, serve_hit
+-- replays them as one text/event-stream body.
+--
+local sse = require("apisix.plugins.ai-transport.sse")
+local protocols = require("apisix.plugins.ai-protocols")
+
+local pcall = pcall
+local require = require
+local type = type
+
+local _M = {}
+
+-- Replay format tag stored on every cache entry (L1 envelope and L2 doc).
+_M.FORMAT_JSON = "json" -- single-shot application/json body
+_M.FORMAT_SSE = "sse" -- raw text/event-stream frames incl. terminal event
+
+
+-- Replayable: single-shot (no framing) or SSE. Binary framings
+-- (aws-eventstream) are not.
+function _M.capturable(ctx)
+ local framing = ctx.ai_stream_framing
+ return not framing or framing == "sse"
+end
+
+
+-- Access-time analog of capturable(): predicts from the picked provider
+-- whether a stream's framing is replayable, so access can skip the lookup.
+function _M.provider_capturable(instance)
+ local ok, provider = pcall(require,
+ "apisix.plugins.ai-providers." ..
instance.provider)
+ if not ok or type(provider) ~= "table" then
+ return true
+ end
+ local framing = provider.streaming_framing
+ return not framing or framing == "sse"
+end
+
+
+-- True when the buffer ends, at a frame boundary, with the client protocol's
+-- terminal event. Deliberately NOT ctx.var.llm_request_done, which is also set
+-- on aborts. The boundary check guards truncation: sse.decode treats a
trailing
+-- partial frame as a complete event, so a cut-off buffer could otherwise pass.
+function _M.stream_completed(ctx, body)
+ if not body or body == "" then
+ return false
+ end
+ local proto = ctx.ai_client_protocol and
protocols.get(ctx.ai_client_protocol)
+ if not (proto and proto.is_done_event) then
+ return false
+ end
+ local tail = body:sub(-256)
+ if not (tail:find("\n\n%s*$") or tail:find("\r\n\r\n%s*$")) then
+ return false
+ end
+ local events = sse.decode(body)
+ local last = events[#events]
+ return last ~= nil and proto.is_done_event(last) == true
+end
+
+
+local function looks_like_sse(body)
+ return body:find("^%s*data:") or body:find("^%s*event:")
+ or body:find("^%s*:") or body:find("^%s*id:") or
body:find("^%s*retry:")
+end
+
+
+-- Format tag to store for a MISS capture, or nil when it must not be cached
+-- (incomplete stream, SSE bytes without framing, non-SSE framing).
+function _M.capture_format(ctx, body)
+ local framing = ctx.ai_stream_framing
+ if not framing then
+ if looks_like_sse(body) then
+ return nil
+ end
+ return _M.FORMAT_JSON
+ end
+ if framing ~= "sse" or not _M.stream_completed(ctx, body) then
+ return nil
+ end
+ return _M.FORMAT_SSE
+end
+
+
+return _M
diff --git a/apisix/plugins/ai-cache/vector-search/redis.lua
b/apisix/plugins/ai-cache/vector-search/redis.lua
index 9b2adeaf7..b50cba17b 100644
--- a/apisix/plugins/ai-cache/vector-search/redis.lua
+++ b/apisix/plugins/ai-cache/vector-search/redis.lua
@@ -74,7 +74,8 @@ function _M.upsert(red, doc_key, fields, ttl)
"partition", fields.partition,
"embedding", fields.embedding,
"response", fields.response,
- "created_at", fields.created_at)
+ "created_at", fields.created_at,
+ "format", fields.format)
red:expire(doc_key, ttl)
local res, err = red:commit_pipeline()
if not res then
@@ -89,14 +90,14 @@ function _M.upsert(red, doc_key, fields, ttl)
end
--- Returns the nearest hit { distance, response, created_at } or nil (no err)
on
--- an empty result set.
+-- Returns the nearest hit { distance, response, created_at, format } or nil
(no
+-- err) on an empty result set.
function _M.knn_search(red, target, index, partition, vec, top_k)
local query = "(@partition:{" .. partition .. "})=>[KNN " .. top_k ..
" @embedding $vec AS __score]"
local res, err = red[ "FT.SEARCH" ](red, index, query,
"PARAMS", 2, "vec", _M.pack_float32(vec),
- "RETURN", 3, "__score", "response", "created_at",
+ "RETURN", 4, "__score", "response", "created_at", "format",
"SORTBY", "__score",
"DIALECT", 2)
if not res then
@@ -121,6 +122,8 @@ function _M.knn_search(red, target, index, partition, vec,
top_k)
hit.response = v
elseif k == "created_at" then
hit.created_at = tonumber(v)
+ elseif k == "format" then
+ hit.format = v
end
end
if not hit.response or not hit.distance then
diff --git a/apisix/plugins/ai-providers/base.lua
b/apisix/plugins/ai-providers/base.lua
index 50b56ca2a..a992defba 100644
--- a/apisix/plugins/ai-providers/base.lua
+++ b/apisix/plugins/ai-providers/base.lua
@@ -456,6 +456,14 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
if not framing then
return 500, "unknown streaming framing: " ..
tostring(self.streaming_framing)
end
+ -- expose the wire framing so response-observing plugins (e.g. ai-cache)
+ -- can classify the stream without re-sniffing the content-type
+ ctx.ai_stream_framing = self.streaming_framing or "sse"
+ -- clear a previous attempt's abort flag: re-entry only happens when that
+ -- attempt emitted no output (with headers sent, the retry dies earlier in
+ -- core.response.set_header), so the flag is stale, not protective
+ ctx.ai_stream_aborted = nil
+ ngx.status = res.status
local body_reader = res.body_reader
local contents = {}
local sse_state = { is_first = true }
@@ -524,6 +532,7 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
local function abort_on_disconnect(flush_err)
core.log.info("client disconnected during AI streaming, ",
"aborting upstream read: ", flush_err)
+ ctx.ai_stream_aborted = "client_disconnect"
if flush_thread then
ngx.thread.kill(flush_thread)
flush_thread = nil
@@ -549,6 +558,7 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
local chunk, err = body_reader()
if err then
+ ctx.ai_stream_aborted = "read_error"
ctx.var.apisix_upstream_response_time = math.floor(
(ngx_now() - ctx.llm_request_start_time) * 1000)
core.log.warn("failed to read response chunk: ", err)
@@ -724,6 +734,7 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
limit_hit = "max_response_bytes"
end
if limit_hit then
+ ctx.ai_stream_aborted = limit_hit
if flush_thread then
ngx.thread.kill(flush_thread)
flush_thread = nil
diff --git a/docs/en/latest/plugins/ai-cache.md
b/docs/en/latest/plugins/ai-cache.md
index d16a9a57f..907b8d177 100644
--- a/docs/en/latest/plugins/ai-cache.md
+++ b/docs/en/latest/plugins/ai-cache.md
@@ -47,6 +47,14 @@ This Plugin supports two cache layers:
The `ai-cache` Plugin must be used with the [`ai-proxy`](./ai-proxy.md) or
[`ai-proxy-multi`](./ai-proxy-multi.md) Plugin.
+### Streaming
+
+Streaming (SSE) responses are cached and replayed. A streamed response is
written to the cache only after it completes — that is, the terminal event for
the client protocol is received (`data: [DONE]` for OpenAI, `message_stop` for
Anthropic, `response.completed` for OpenAI Responses). A stream that is
interrupted (client disconnect, or the `ai-proxy` `max_stream_duration_ms` /
`max_response_bytes` limits) is never cached, so partial responses are never
served. On a hit, the stored respo [...]
+
+Streaming and non-streaming requests for the same prompt are cached as
**separate** entries at both layers, so a streaming client always receives a
stream and a non-streaming client always receives a single JSON response. This
applies whether streaming is requested by the client (`"stream": true`) or
forced by the route via `options.stream`.
+
+Limitations: binary streaming formats without an SSE terminal event (for
example Bedrock ConverseStream) are not cached. Replay is immediate (the full
stored response is sent at once) rather than re-timed token-by-token.
+
:::note
By default the cache is isolated per route, so two routes never serve each
other's entries even when they see the same protocol, model and messages. Set
`cache_key.share_across_routes` to `true` to share one cache space across
routes.
diff --git a/docs/zh/latest/plugins/ai-cache.md
b/docs/zh/latest/plugins/ai-cache.md
index 16f3c2b7c..5c9e51234 100644
--- a/docs/zh/latest/plugins/ai-cache.md
+++ b/docs/zh/latest/plugins/ai-cache.md
@@ -47,6 +47,14 @@ import TabItem from '@theme/TabItem';
`ai-cache` 插件必须与 [`ai-proxy`](./ai-proxy.md) 或
[`ai-proxy-multi`](./ai-proxy-multi.md) 插件一起使用。
+### 流式响应
+
+插件支持缓存并回放流式（SSE）响应。流式响应仅在**完成后**才写入缓存，即接收到客户端协议对应的终止事件（OpenAI 为 `data:
[DONE]`，Anthropic 为 `message_stop`，OpenAI Responses 为
`response.completed`）。被中断的流（客户端断开连接，或触发 `ai-proxy` 的 `max_stream_duration_ms` /
`max_response_bytes` 限制）不会被缓存，因此不会回放不完整的响应。命中缓存时，存储的响应会作为单个 `text/event-stream`
响应体完整回放，并保留其终止事件。
+
+对于相同的提示词，流式请求与非流式请求会在两个缓存层中分别存储为**独立**的条目，因此流式客户端始终收到流式响应，非流式客户端始终收到单个 JSON
响应。无论流式是由客户端请求（`"stream": true`）还是由路由通过 `options.stream` 强制开启，均是如此。
+
+限制：不含 SSE 终止事件的二进制流式格式（例如 Bedrock
ConverseStream）不会被缓存；回放是即时的（一次性发送完整的存储响应），而非按 token 重新计时逐个发送。
+
:::note
默认情况下缓存按路由隔离，因此即使两个路由看到相同的协议、模型与消息，也不会相互返回对方的缓存条目。将
`cache_key.share_across_routes` 设为 `true` 可让多个路由共享同一个缓存空间。
diff --git a/t/lib/ai_cache_mock.lua b/t/lib/ai_cache_mock.lua
index 733014bfc..18de8bbce 100644
--- a/t/lib/ai_cache_mock.lua
+++ b/t/lib/ai_cache_mock.lua
@@ -101,6 +101,21 @@ function _M.embeddings_azure()
end
+-- First hit: 200 + SSE headers then stall (read timeout, no body); later hits
serve the fixture.
+local flaky_hits = 0
+function _M.chat_flaky_once()
+ flaky_hits = flaky_hits + 1
+ if flaky_hits == 1 then
+ ngx.header["Content-Type"] = "text/event-stream"
+ ngx.send_headers()
+ ngx.flush(true)
+ ngx.sleep(2)
+ return
+ end
+ fixture_loader.dispatch()
+end
+
+
-- always 5xx: drives the embedding-provider fail-open path.
function _M.broken()
ngx.status = 500
diff --git a/t/plugin/ai-cache-semantic.t b/t/plugin/ai-cache-semantic.t
index 4de4e50d6..db2a15d93 100644
--- a/t/plugin/ai-cache-semantic.t
+++ b/t/plugin/ai-cache-semantic.t
@@ -334,30 +334,31 @@ qr/property "message_countback" validation failed:
expected 0 to be at least 1/
-=== TEST 13: context_fingerprint ignores message TEXT but reacts to
model/params (key.lua unit)
+=== TEST 13: partition ignores message TEXT but reacts to model/params
(key.lua unit)
--- config
location /t {
content_by_lua_block {
local key = require("apisix.plugins.ai-cache.key")
- -- the context fingerprint deduplicates by the EFFECTIVE request
context
+ -- the partition deduplicates by the EFFECTIVE request context
-- (provider/model/params/instance) with the message wording
removed, so a
- -- paraphrase collapses to one fingerprint while a parameter
change does not.
+ -- paraphrase collapses to one partition while a parameter change
does not.
local function ctx()
return { ai_client_protocol = "openai-chat",
picked_ai_instance = { provider = "openai",
options = { model = "gpt-4o"
}, override = {} },
- var = {} }
+ var = { route_id = "1" } }
end
- local a = key.context_fingerprint(ctx(), { model = "gpt-4o",
temperature = 0.2,
- messages = {{ role = "user", content = "how do I return an
item?" }} })
- local b = key.context_fingerprint(ctx(), { model = "gpt-4o",
temperature = 0.2,
- messages = {{ role = "user", content = "what is the return
policy?" }} })
- local c = key.context_fingerprint(ctx(), { model = "gpt-4o",
temperature = 0.9,
- messages = {{ role = "user", content = "how do I return an
item?" }} })
-
- assert(a == b, "differently-worded prompts must share one context
fingerprint")
- assert(a ~= c, "a parameter change (temperature) must flip the
context fingerprint")
+ local conf = { cache_key = {} }
+ local a = key.partition(conf, ctx(), { model = "gpt-4o",
temperature = 0.2,
+ messages = {{ role = "user", content = "how do I return an
item?" }} }, nil)
+ local b = key.partition(conf, ctx(), { model = "gpt-4o",
temperature = 0.2,
+ messages = {{ role = "user", content = "what is the return
policy?" }} }, nil)
+ local c = key.partition(conf, ctx(), { model = "gpt-4o",
temperature = 0.9,
+ messages = {{ role = "user", content = "how do I return an
item?" }} }, nil)
+
+ assert(a == b, "differently-worded prompts must share one
partition")
+ assert(a ~= c, "a parameter change (temperature) must flip the
partition")
ngx.say("passed")
}
}
@@ -619,10 +620,10 @@ passed
assert(vs.upsert(red, "ut-knn:l2:p1:near",
{ partition = "p1", embedding = vs.pack_float32({ 1.0, 0.0,
0.0 }),
- response = [[{"answer":"NEAR"}]], created_at = 100 }, 600))
+ response = [[{"answer":"NEAR"}]], created_at = 100, format =
"json" }, 600))
assert(vs.upsert(red, "ut-knn:l2:p1:far",
{ partition = "p1", embedding = vs.pack_float32({ 0.0, 1.0,
0.0 }),
- response = [[{"answer":"FAR"}]], created_at = 100 }, 600))
+ response = [[{"answer":"FAR"}]], created_at = 100, format =
"json" }, 600))
local hit, err = vs.knn_search(red, tgt, "ut-knn:idx:3", "p1", {
0.99, 0.01, 0.0 }, 1)
if not hit then ngx.say("no-hit:", err or ""); return end
@@ -672,10 +673,10 @@ clean-miss
-- two docs with the SAME vector but different partition tags
assert(vs.upsert(red, "ut-part:l2:p1:d1",
{ partition = "p1", embedding = vs.pack_float32({ 1.0, 0.0,
0.0 }),
- response = [[{"answer":"P1"}]], created_at = 100 }, 600))
+ response = [[{"answer":"P1"}]], created_at = 100, format =
"json" }, 600))
assert(vs.upsert(red, "ut-part:l2:p2:d2",
{ partition = "p2", embedding = vs.pack_float32({ 1.0, 0.0,
0.0 }),
- response = [[{"answer":"P2"}]], created_at = 100 }, 600))
+ response = [[{"answer":"P2"}]], created_at = 100, format =
"json" }, 600))
local hit = vs.knn_search(red, tgt, "ut-part:idx:3", "p1", { 1, 0,
0 }, 1)
ngx.say(hit and hit.response or "no-hit")
diff --git a/t/plugin/ai-cache-streaming.t b/t/plugin/ai-cache-streaming.t
new file mode 100644
index 000000000..6a18e9b02
--- /dev/null
+++ b/t/plugin/ai-cache-streaming.t
@@ -0,0 +1,909 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+use t::APISIX 'no_plan';
+
+log_level("info");
+repeat_each(1);
+no_long_string();
+no_root_location();
+
+add_block_preprocessor(sub {
+ my ($block) = @_;
+
+ if (!defined $block->request) {
+ $block->set_value("request", "GET /t");
+ }
+
+ my $user_yaml_config = <<_EOC_;
+plugins:
+ - ai-proxy
+ - ai-cache
+_EOC_
+ if (!defined $block->extra_yaml_config) {
+ $block->set_value("extra_yaml_config", $user_yaml_config);
+ }
+
+ my $http_config = $block->http_config // <<_EOC_;
+ server {
+ listen 6724;
+ default_type 'text/event-stream';
+
+ location /v1/embeddings {
+ content_by_lua_block { require("lib.ai_cache_mock").embeddings() }
+ }
+
+ location /v1/chat/completions-truncated {
+ content_by_lua_block {
+ ngx.header["Content-Type"] = "text/event-stream"
+ ngx.print('data:
{"id":"1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"Hel"},"finish_reason":null}]}\\n\\n')
+ ngx.flush(true)
+ }
+ }
+
+ location /v1/chat/completions-runaway {
+ content_by_lua_block {
+ ngx.header["Content-Type"] = "text/event-stream"
+ for i = 1, 40 do
+ ngx.print('data:
{"id":"1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"tokentokentokentokentoken"},"finish_reason":null}]}\\n\\n')
+ ngx.flush(true)
+ end
+ ngx.print('data: [DONE]\\n\\n')
+ ngx.flush(true)
+ }
+ }
+
+ # An SSE stream carried under a non-success status (HTTP 400) that
still
+ # ends with a valid [DONE] sentinel: ai-proxy's 429/5xx gate does not
catch
+ # 400, so without status propagation this would stream as 200 and get
+ # cached as a successful HIT.
+ location /v1/chat/completions-error-sse {
+ content_by_lua_block {
+ ngx.status = 400
+ ngx.header["Content-Type"] = "text/event-stream"
+ ngx.print('data: {"error":{"message":"bad request"}}\\n\\n')
+ ngx.print('data: [DONE]\\n\\n')
+ ngx.flush(true)
+ }
+ }
+
+ location /v1/chat/completions-flaky-once {
+ content_by_lua_block {
require("lib.ai_cache_mock").chat_flaky_once() }
+ }
+ }
+_EOC_
+ $block->set_value("http_config", $http_config);
+});
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: folding stream into the fingerprint isolates stream from non-stream
+--- config
+ location /t {
+ content_by_lua_block {
+ local key = require("apisix.plugins.ai-cache.key")
+ local ctx = {
+ picked_ai_instance = { provider = "openai", options = { model
= "gpt-4" } },
+ ai_client_protocol = "openai-chat",
+ var = { route_id = "1" },
+ }
+ local base = { messages = {{ role = "user", content = "hi" }},
model = "gpt-4" }
+ local stream = { messages = {{ role = "user", content = "hi" }},
model = "gpt-4", stream = true }
+ local fp_a = key.fingerprint(ctx, base)
+ local fp_b = key.fingerprint(ctx, stream)
+ local conf = { cache_key = {} }
+ local pt_a = key.partition(conf, ctx, base, nil)
+ local pt_b = key.partition(conf, ctx, stream, nil)
+ ngx.say((fp_a ~= fp_b) and (pt_a ~= pt_b) and "ISOLATED" or
"COLLISION")
+ }
+ }
+--- response_body
+ISOLATED
+
+
+
+=== TEST 2: stream_completed is protocol-aware (openai [DONE], anthropic
message_stop) and immune to a [DONE] substring
+--- config
+ location /t {
+ content_by_lua_block {
+ local stream = require("apisix.plugins.ai-cache.stream")
+ local octx = { ai_client_protocol = "openai-chat", var = {} }
+ local actx = { ai_client_protocol = "anthropic-messages", var = {}
}
+ local complete = 'data:
{"choices":[{"delta":{"content":"Hi"}}]}\n\n' .. 'data: [DONE]\n\n'
+ local truncated = 'data:
{"choices":[{"delta":{"content":"Hi"}}]}\n\n'
+ -- a "[DONE]" substring inside content must NOT count as completion
+ local fake_done = 'data:
{"choices":[{"delta":{"content":"[DONE]"}}]}\n\n'
+ -- anthropic's terminal sentinel is the message_stop event, not
[DONE]
+ local anthropic_done = 'event: message_stop\ndata: {}\n\n'
+ local anthropic_truncated = 'event: message_start\ndata: {}\n\n'
+ -- a terminal frame cut off mid-data (no closing blank line) must
NOT
+ -- count as complete, even though its event TYPE parses as
message_stop
+ local anthropic_partial = 'event: message_start\ndata: {}\n\n'
+ .. 'event: message_stop\ndata: {"ty'
+ -- spec-legal comment/heartbeat frames AFTER the terminal event
must
+ -- not hide it (upstreams/proxies that ping after [DONE])
+ local keepalive_after_done = complete .. ': keepalive\n\n'
+ -- ...but a comment truncated mid-write is not a frame boundary
+ local keepalive_truncated = complete .. ': keepal'
+ ngx.say(table.concat({
+ tostring(stream.stream_completed(octx, complete)),
+ tostring(stream.stream_completed(octx, truncated)),
+ tostring(stream.stream_completed(octx, fake_done)),
+ tostring(stream.stream_completed(actx, anthropic_done)),
+ tostring(stream.stream_completed(actx, anthropic_truncated)),
+ tostring(stream.stream_completed(actx, anthropic_partial)),
+ tostring(stream.stream_completed(octx, keepalive_after_done)),
+ tostring(stream.stream_completed(octx, keepalive_truncated)),
+ }, ","))
+ }
+ }
+--- response_body
+true,false,false,true,false,false,true,false
+
+
+
+=== TEST 3: capture_format tags a capture by wire framing and completeness
+--- config
+ location /t {
+ content_by_lua_block {
+ local stream = require("apisix.plugins.ai-cache.stream")
+ -- framing is stamped by ai-providers parse_streaming_response;
+ -- absent means the response was a single-shot (non-streaming) body
+ local plain = { var = {} }
+ local sse_ctx = { ai_stream_framing = "sse",
+ ai_client_protocol = "openai-chat", var = {} }
+ local bin_ctx = { ai_stream_framing = "aws-eventstream",
+ ai_client_protocol = "bedrock-converse", var =
{} }
+ local done = 'data: {"choices":[]}\n\ndata: [DONE]\n\n'
+ ngx.say(table.concat({
+ stream.capture_format(plain, '{"id":"1"}'), -- json
+ stream.capture_format(sse_ctx, done), -- sse
+ tostring(stream.capture_format(sse_ctx, 'data: {}\n\n')), --
nil: incomplete
+ tostring(stream.capture_format(bin_ctx, done)), -- nil:
binary framing
+ tostring(stream.capture_format(plain, done)), -- nil:
mislabeled sse
+ tostring(stream.capture_format(plain, ': ping\ndata:
{}\n\n')), -- nil: comment first line
+ tostring(stream.capturable(plain)), -- true
+ tostring(stream.capturable(sse_ctx)), -- true
+ tostring(stream.capturable(bin_ctx)), -- false
+ -- access-time prediction from the picked provider
+ tostring(stream.provider_capturable({provider = "openai"})),
-- true
+ tostring(stream.provider_capturable({provider = "bedrock"})),
-- false
+ }, ","))
+ }
+ }
+--- response_body
+json,sse,nil,nil,nil,nil,true,true,false,true,false
+
+
+
+=== TEST 4: set a streaming route (ai-proxy stream + ai-cache), pointed at the
:1980 SSE fixture
+--- config
+ location /t {
+ content_by_lua_block {
+ require("lib.test_redis").flush_port("127.0.0.1", 6379)
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, [[{
+ "uri": "/stream",
+ "plugins": {
+ "ai-proxy": {
+ "provider": "openai",
+ "auth": { "header": { "Authorization": "Bearer test" }
},
+ "options": { "model": "gpt-4", "stream": true },
+ "override": { "endpoint": "http://127.0.0.1:1980"; }
+ },
+ "ai-cache": { "redis_host": "127.0.0.1", "redis_port":
6379 }
+ }
+ }]])
+ if code >= 300 then ngx.status = code end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 5: a cold streaming request is a MISS and is streamed from upstream
+--- request
+POST /stream
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- more_headers
+X-AI-Fixture: openai/chat-streaming.sse
+--- response_headers_like
+X-AI-Cache-Status: MISS
+Content-Type: text/event-stream
+--- response_body_like
+data: \[DONE\]
+--- wait: 0.5
+
+
+
+=== TEST 6: an identical streaming request is a HIT, replayed as a valid
text/event-stream
+--- request
+POST /stream
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- more_headers
+X-AI-Fixture: openai/chat-streaming.sse
+--- response_headers_like
+X-AI-Cache-Status: HIT
+X-AI-Cache-Age: \d+
+Content-Type: text/event-stream
+--- response_body_like
+data: \[DONE\]
+
+
+
+=== TEST 7: set a route whose upstream truncates the stream (no terminal
[DONE])
+--- config
+ location /t {
+ content_by_lua_block {
+ require("lib.test_redis").flush_port("127.0.0.1", 6379)
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, [[{
+ "uri": "/stream-trunc",
+ "plugins": {
+ "ai-proxy": {
+ "provider": "openai",
+ "auth": { "header": { "Authorization": "Bearer test" }
},
+ "options": { "model": "gpt-4", "stream": true },
+ "override": { "endpoint":
"http://127.0.0.1:6724/v1/chat/completions-truncated"; }
+ },
+ "ai-cache": { "redis_host": "127.0.0.1", "redis_port":
6379 }
+ }
+ }]])
+ if code >= 300 then ngx.status = code end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 8: the truncated stream is a MISS and is NOT written back (no
terminal sentinel)
+--- request
+POST /stream-trunc
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- response_headers_like
+X-AI-Cache-Status: MISS
+--- wait: 0.5
+
+
+
+=== TEST 9: an identical request is STILL a MISS -- the truncated stream was
never cached
+--- request
+POST /stream-trunc
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- response_headers_like
+X-AI-Cache-Status: MISS
+
+
+
+=== TEST 10: set a route that FORCES streaming via options.stream (client body
has none)
+--- config
+ location /t {
+ content_by_lua_block {
+ require("lib.test_redis").flush_port("127.0.0.1", 6379)
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, [[{
+ "uri": "/forced-stream",
+ "plugins": {
+ "ai-proxy": {
+ "provider": "openai",
+ "auth": { "header": { "Authorization": "Bearer test" }
},
+ "options": { "model": "gpt-4", "stream": true },
+ "override": { "endpoint": "http://127.0.0.1:1980"; }
+ },
+ "ai-cache": { "redis_host": "127.0.0.1", "redis_port":
6379 }
+ }
+ }]])
+ if code >= 300 then ngx.status = code end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 11: options.stream makes the response SSE though request_type is
ai_chat -- MISS, streamed
+--- request
+POST /forced-stream
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4"}
+--- more_headers
+X-AI-Fixture: openai/chat-streaming.sse
+--- response_headers_like
+X-AI-Cache-Status: MISS
+Content-Type: text/event-stream
+--- wait: 0.5
+
+
+
+=== TEST 12: the forced-stream entry replays as text/event-stream (format
follows the response, not request_type)
+--- request
+POST /forced-stream
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4"}
+--- more_headers
+X-AI-Fixture: openai/chat-streaming.sse
+--- response_headers_like
+X-AI-Cache-Status: HIT
+Content-Type: text/event-stream
+--- response_body_like
+data: \[DONE\]
+
+
+
+=== TEST 13: set a non-streaming and a streaming route that share one cache
space
+--- config
+ location /t {
+ content_by_lua_block {
+ require("lib.test_redis").flush_port("127.0.0.1", 6379)
+ local t = require("lib.test_admin").test
+ local code = t('/apisix/admin/routes/1', ngx.HTTP_PUT, [[{
+ "uri": "/iso-json", "plugins": {
+ "ai-proxy": { "provider": "openai",
+ "auth": { "header": { "Authorization": "Bearer test" }
},
+ "options": { "model": "gpt-4" },
+ "override": { "endpoint": "http://127.0.0.1:1980"; } },
+ "ai-cache": { "redis_host": "127.0.0.1", "redis_port":
6379,
+ "cache_key": { "share_across_routes": true } } } }]])
+ if code >= 300 then ngx.status = code; ngx.say(code); return end
+ code = t('/apisix/admin/routes/2', ngx.HTTP_PUT, [[{
+ "uri": "/iso-sse", "plugins": {
+ "ai-proxy": { "provider": "openai",
+ "auth": { "header": { "Authorization": "Bearer test" }
},
+ "options": { "model": "gpt-4", "stream": true },
+ "override": { "endpoint": "http://127.0.0.1:1980"; } },
+ "ai-cache": { "redis_host": "127.0.0.1", "redis_port":
6379,
+ "cache_key": { "share_across_routes": true } } } }]])
+ ngx.say(code < 300 and "passed" or code)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 14: prime the non-streaming (JSON) entry
+--- request
+POST /iso-json
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4"}
+--- more_headers
+X-AI-Fixture: openai/chat-basic.json
+--- response_headers_like
+X-AI-Cache-Status: MISS
+--- wait: 0.5
+
+
+
+=== TEST 15: the SAME prompt as a stream is a MISS (never serves the JSON
entry) and is SSE
+--- request
+POST /iso-sse
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- more_headers
+X-AI-Fixture: openai/chat-streaming.sse
+--- response_headers_like
+X-AI-Cache-Status: MISS
+Content-Type: text/event-stream
+--- wait: 0.5
+
+
+
+=== TEST 16: reverse isolation -- flush and reuse the shared routes from TEST
13
+--- config
+ location /t {
+ content_by_lua_block {
+ require("lib.test_redis").flush_port("127.0.0.1", 6379)
+ ngx.say("flushed")
+ }
+ }
+--- response_body
+flushed
+
+
+
+=== TEST 17: prime the streaming (SSE) entry
+--- request
+POST /iso-sse
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- more_headers
+X-AI-Fixture: openai/chat-streaming.sse
+--- response_headers_like
+X-AI-Cache-Status: MISS
+--- wait: 0.5
+
+
+
+=== TEST 18: the SAME prompt non-streamed is a MISS (never serves the SSE
entry) and is JSON
+--- request
+POST /iso-json
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4"}
+--- more_headers
+X-AI-Fixture: openai/chat-basic.json
+--- response_headers_like
+X-AI-Cache-Status: MISS
+Content-Type: application/json
+
+
+
+=== TEST 19: set a runaway streaming route with a low max_response_bytes
+--- config
+ location /t {
+ content_by_lua_block {
+ require("lib.test_redis").flush_port("127.0.0.1", 6379)
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, [[{
+ "uri": "/stream-runaway",
+ "plugins": {
+ "ai-proxy": {
+ "provider": "openai",
+ "auth": { "header": { "Authorization": "Bearer test" }
},
+ "options": { "model": "gpt-4", "stream": true },
+ "max_response_bytes": 512,
+ "override": { "endpoint":
"http://127.0.0.1:6724/v1/chat/completions-runaway"; }
+ },
+ "ai-cache": { "redis_host": "127.0.0.1", "redis_port":
6379 }
+ }
+ }]])
+ if code >= 300 then ngx.status = code end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 20: max_response_bytes aborts the stream (llm_request_done=true)
before [DONE]
+--- request
+POST /stream-runaway
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- error_log
+aborting AI stream: max_response_bytes exceeded
+--- wait: 0.5
+
+
+
+=== TEST 21: the aborted stream was NOT cached -- an identical request is
STILL a MISS
+--- request
+POST /stream-runaway
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- response_headers_like
+X-AI-Cache-Status: MISS
+
+
+
+=== TEST 22: set a semantic (exact+semantic) streaming route
+--- config
+ location /t {
+ content_by_lua_block {
+ require("lib.test_redis").flush_port("127.0.0.1", 6379)
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, [[{
+ "uri": "/sse-sem", "plugins": {
+ "ai-proxy": { "provider": "openai",
+ "auth": { "header": { "Authorization": "Bearer
test-key" } },
+ "options": { "model": "gpt-4o", "stream": true },
+ "override": { "endpoint": "http://127.0.0.1:1980"; } },
+ "ai-cache": { "redis_host": "127.0.0.1", "redis_port":
6379,
+ "layers": ["exact","semantic"],
+ "semantic": {
+ "similarity_threshold": 0.9,
+ "embedding": { "openai": {
+ "endpoint":
"http://127.0.0.1:6724/v1/embeddings";,
+ "model": "text-embedding-3-small", "api_key":
"test-key" } },
+ "vector_search": { "redis": {} } } } } }]])
+ if code >= 300 then ngx.status = code end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 23: a cold streaming prompt is a semantic MISS
+--- request
+POST /sse-sem
+{"model":"gpt-4o","messages":[{"role":"user","content":"What is the capital of
France?"}],"stream":true}
+--- more_headers
+X-AI-Fixture: openai/chat-streaming.sse
+--- response_headers_like
+X-AI-Cache-Status: MISS
+--- wait: 0.6
+
+
+
+=== TEST 24: a paraphrased streaming prompt is a semantic (L2) HIT, replayed
as SSE
+--- request
+POST /sse-sem
+{"model":"gpt-4o","messages":[{"role":"user","content":"What is the capital
city of France?"}],"stream":true}
+--- more_headers
+X-AI-Fixture: openai/chat-streaming.sse
+--- response_headers_like
+X-AI-Cache-Status: HIT
+X-AI-Cache-Similarity: 0\.92\d\d
+Content-Type: text/event-stream
+--- response_body_like
+data: \[DONE\]
+
+
+
+=== TEST 25: set a streaming and a non-streaming semantic route over one
shared index
+--- config
+ location /t {
+ content_by_lua_block {
+ require("lib.test_redis").flush_port("127.0.0.1", 6379)
+ local t = require("lib.test_admin").test
+ local function sem(uri, stream_opt)
+ return [[{
+ "uri": "]] .. uri .. [[", "plugins": {
+ "ai-proxy": { "provider": "openai",
+ "auth": { "header": { "Authorization": "Bearer
test-key" } },
+ "options": { "model": "gpt-4o"]] .. stream_opt ..
[[ },
+ "override": { "endpoint": "http://127.0.0.1:1980";
} },
+ "ai-cache": { "redis_host": "127.0.0.1", "redis_port":
6379,
+ "cache_key": { "share_across_routes": true },
+ "layers": ["exact","semantic"],
+ "semantic": {
+ "similarity_threshold": 0.9,
+ "embedding": { "openai": {
+ "endpoint":
"http://127.0.0.1:6724/v1/embeddings";,
+ "model": "text-embedding-3-small",
"api_key": "test-key" } },
+ "vector_search": { "redis": { "index":
"ai-cache-iso" } } } } } }]]
+ end
+ local code = t('/apisix/admin/routes/1', ngx.HTTP_PUT,
sem("/sse-sem2", ', "stream": true'))
+ if code >= 300 then ngx.status = code; ngx.say(code); return end
+ code = t('/apisix/admin/routes/2', ngx.HTTP_PUT, sem("/json-sem2",
""))
+ ngx.say(code < 300 and "passed" or code)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 26: prime a STREAM L2 doc
+--- request
+POST /sse-sem2
+{"model":"gpt-4o","messages":[{"role":"user","content":"What is the capital of
France?"}],"stream":true}
+--- more_headers
+X-AI-Fixture: openai/chat-streaming.sse
+--- response_headers_like
+X-AI-Cache-Status: MISS
+--- wait: 0.6
+
+
+
+=== TEST 27: a non-stream paraphrase MISSES (the stream doc is in a different
partition)
+--- request
+POST /json-sem2
+{"model":"gpt-4o","messages":[{"role":"user","content":"What is the capital
city of France?"}]}
+--- more_headers
+X-AI-Fixture: openai/chat-basic.json
+--- response_headers_like
+X-AI-Cache-Status: MISS
+--- wait: 0.6
+
+
+
+=== TEST 28: a non-stream paraphrase of TEST 27 HITs L2 -- L2 works for
non-stream, stream doc stayed isolated
+--- request
+POST /json-sem2
+{"model":"gpt-4o","messages":[{"role":"user","content":"What is the capital of
France?"}]}
+--- more_headers
+X-AI-Fixture: openai/chat-basic.json
+--- response_headers_like
+X-AI-Cache-Status: HIT
+Content-Type: application/json
+
+
+
+=== TEST 29: buffered/one-write upstream + low max_response_bytes (limit trips
AFTER [DONE] is buffered)
+--- config
+ location /t {
+ content_by_lua_block {
+ require("lib.test_redis").flush_port("127.0.0.1", 6379)
+ local t = require("lib.test_admin").test
+ -- The shared :1980 loader sends the whole .sse fixture (826 bytes,
+ -- ending with [DONE]) in ONE write, so the terminal sentinel is
+ -- already in the captured buffer when max_response_bytes (256)
trips.
+ -- stream_completed() alone would treat this as a cacheable
complete
+ -- stream; ctx.ai_stream_aborted must veto the write.
+ local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, [[{
+ "uri": "/stream-onewrite",
+ "plugins": {
+ "ai-proxy": {
+ "provider": "openai",
+ "auth": { "header": { "Authorization": "Bearer test" }
},
+ "options": { "model": "gpt-4", "stream": true },
+ "max_response_bytes": 256,
+ "override": { "endpoint": "http://127.0.0.1:1980"; }
+ },
+ "ai-cache": { "redis_host": "127.0.0.1", "redis_port":
6379 }
+ }
+ }]])
+ if code >= 300 then ngx.status = code end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 30: the one-write stream carries [DONE] yet max_response_bytes still
aborts it
+--- request
+POST /stream-onewrite
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- more_headers
+X-AI-Fixture: openai/chat-streaming.sse
+--- error_log
+aborting AI stream: max_response_bytes exceeded
+--- wait: 0.5
+
+
+
+=== TEST 31: the limit-aborted stream was NOT cached despite the buffered
[DONE] -- identical request STILL a MISS
+--- request
+POST /stream-onewrite
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- more_headers
+X-AI-Fixture: openai/chat-streaming.sse
+--- response_headers_like
+X-AI-Cache-Status: MISS
+--- wait: 0.5
+
+
+
+=== TEST 32: stream folds by `== true` (mirrors is_streaming);
false/omitted/null/0 share ONE key, only true isolates
+--- config
+ location /t {
+ content_by_lua_block {
+ local key = require("apisix.plugins.ai-cache.key")
+ local null = require("cjson").null -- what cjson.decode gives
for JSON null
+ local ctx = {
+ picked_ai_instance = { provider = "openai", options = { model
= "gpt-4" } },
+ ai_client_protocol = "openai-chat",
+ var = { route_id = "1" },
+ }
+ local msgs = {{ role = "user", content = "hi" }}
+ local conf = { cache_key = {} }
+ local function fp(stream_val)
+ local body = { messages = msgs, model = "gpt-4" }
+ if stream_val ~= nil then body.stream = stream_val end
+ return key.fingerprint(ctx, body)
+ end
+ -- ai-proxy is_streaming does `body.stream == true`, so null
(truthy
+ -- userdata) and 0 (truthy in Lua) are NON-streaming and must not
split
+ -- off from omitted/false, nor collide with a real streaming
request.
+ local base = fp(nil)
+ local non_stream = (fp(false) == base) and (fp(null) == base) and
(fp(0) == base)
+ ngx.say((non_stream and fp(true) ~= base) and "CANONICAL" or
"FRAGMENTED")
+ }
+ }
+--- response_body
+CANONICAL
+
+
+
+=== TEST 33: set a streaming route whose upstream returns HTTP 400 SSE ending
in [DONE]
+--- config
+ location /t {
+ content_by_lua_block {
+ require("lib.test_redis").flush_port("127.0.0.1", 6379)
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, [[{
+ "uri": "/stream-error-sse",
+ "plugins": {
+ "ai-proxy": {
+ "provider": "openai",
+ "auth": { "header": { "Authorization": "Bearer test" }
},
+ "options": { "model": "gpt-4", "stream": true },
+ "override": { "endpoint":
"http://127.0.0.1:6724/v1/chat/completions-error-sse"; }
+ },
+ "ai-cache": { "redis_host": "127.0.0.1", "redis_port":
6379 }
+ }
+ }]])
+ if code >= 300 then ngx.status = code end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 34: the 400 upstream status is propagated (not served as 200) and the
error stream is a MISS
+--- request
+POST /stream-error-sse
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- error_code: 400
+--- response_headers_like
+X-AI-Cache-Status: MISS
+--- wait: 0.5
+
+
+
+=== TEST 35: the 400 error stream was NOT cached despite the [DONE] sentinel
-- identical request STILL a MISS (400)
+--- request
+POST /stream-error-sse
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- error_code: 400
+--- response_headers_like
+X-AI-Cache-Status: MISS
+--- wait: 0.5
+
+
+
+=== TEST 36: set an ai-proxy-multi route on the flaky-once upstream (weight-0
spare keeps the re-pick on the SAME instance)
+--- extra_yaml_config
+plugins:
+ - ai-proxy-multi
+ - ai-cache
+--- config
+ location /t {
+ content_by_lua_block {
+ require("lib.test_redis").flush_port("127.0.0.1", 6379)
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, [[{
+ "uri": "/stream-retry",
+ "plugins": {
+ "ai-proxy-multi": {
+ "fallback_strategy": ["http_5xx"],
+ "timeout": 1000,
+ "ssl_verify": false,
+ "instances": [
+ {"name":"flaky","provider":"openai","weight":1,
+ "auth":{"header":{"Authorization":"Bearer test"}},
+ "options":{"model":"gpt-4","stream":true},
+
"override":{"endpoint":"http://127.0.0.1:6724/v1/chat/completions-flaky-once"}},
+ {"name":"spare","provider":"openai","weight":0,
+ "auth":{"header":{"Authorization":"Bearer test"}},
+ "options":{"model":"gpt-4","stream":true},
+
"override":{"endpoint":"http://127.0.0.1:6724/v1/chat/completions-flaky-once"}}
+ ]
+ },
+ "ai-cache": { "redis_host": "127.0.0.1", "redis_port":
6379 }
+ }
+ }]])
+ if code >= 300 then ngx.status = code end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 37: attempt 1 dies before the first byte, the retry re-picks the SAME
instance and streams to [DONE]
+--- extra_yaml_config
+plugins:
+ - ai-proxy-multi
+ - ai-cache
+--- request
+POST /stream-retry
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- more_headers
+X-AI-Fixture: openai/chat-streaming.sse
+--- response_headers_like
+X-AI-Cache-Status: MISS
+Content-Type: text/event-stream
+--- response_body_like
+data: \[DONE\]
+--- error_log
+failed to read response chunk
+falling back to flaky
+--- timeout: 5
+--- wait: 0.5
+
+
+
+=== TEST 38: the retried stream WAS cached -- attempt 1's stale abort flag no
longer vetoes the write
+--- extra_yaml_config
+plugins:
+ - ai-proxy-multi
+ - ai-cache
+--- request
+POST /stream-retry
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- more_headers
+X-AI-Fixture: openai/chat-streaming.sse
+--- response_headers_like
+X-AI-Cache-Status: HIT
+Content-Type: text/event-stream
+--- response_body_like
+data: \[DONE\]
+
+
+
+=== TEST 39: set a bedrock streaming route (aws-eventstream framing) with
ai-cache
+--- config
+ location /t {
+ content_by_lua_block {
+ require("lib.test_redis").flush_port("127.0.0.1", 6379)
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, [[{
+ "uri": "/bedrock-stream/converse",
+ "plugins": {
+ "ai-proxy": {
+ "provider": "bedrock",
+ "auth": {
+ "aws": {
+ "access_key_id": "AKIAIOSFODNN7EXAMPLE",
+ "secret_access_key":
"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
+ }
+ },
+ "provider_conf": { "region": "us-east-1" },
+ "options": { "model":
"anthropic.claude-3-5-sonnet-20241022-v2:0" },
+ "override": { "endpoint": "http://127.0.0.1:1980"; },
+ "ssl_verify": false
+ },
+ "ai-cache": { "redis_host": "127.0.0.1", "redis_port":
6379 }
+ }
+ }]])
+ if code >= 300 then ngx.status = code end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 40: a binary-framed stream BYPASSes the lookup (could never be
written back) and flows through intact
+--- main_config
+ env AWS_EC2_METADATA_DISABLED=true;
+--- request
+POST /bedrock-stream/converse
+{"stream":true,"messages":[{"role":"user","content":[{"text":"Say hi"}]}]}
+--- error_code: 200
+--- response_headers
+X-AI-Cache-Status: BYPASS
+Content-Type: application/vnd.amazon.eventstream
+--- response_body eval
+qr/messageStart.*contentBlockDelta.*messageStop/s
+--- wait: 0.5
+
+
+
+=== TEST 41: the bypassed stream was never cached -- identical request STILL a
BYPASS
+--- main_config
+ env AWS_EC2_METADATA_DISABLED=true;
+--- request
+POST /bedrock-stream/converse
+{"stream":true,"messages":[{"role":"user","content":[{"text":"Say hi"}]}]}
+--- error_code: 200
+--- response_headers
+X-AI-Cache-Status: BYPASS
+Content-Type: application/vnd.amazon.eventstream
+
+
+
+=== TEST 42: a NON-stream request on the same bedrock route is a MISS -- the
bypass is stream-scoped
+--- main_config
+ env AWS_EC2_METADATA_DISABLED=true;
+--- request
+POST /bedrock-stream/converse
+{"messages":[{"role":"user","content":[{"text":"What is 1+1?"}]}]}
+--- error_code: 200
+--- response_headers
+X-AI-Cache-Status: MISS
+--- response_body eval
+qr/"text"\s*:\s*"Hello!"/
diff --git a/t/plugin/ai-cache.t b/t/plugin/ai-cache.t
index ddc72eb1c..5419d5ef9 100644
--- a/t/plugin/ai-cache.t
+++ b/t/plugin/ai-cache.t
@@ -225,7 +225,7 @@ qr/1 \+ 1 = 2/
assert(fp(inst, other_messages) ~= base, "different messages")
assert(fp(inst, other_model) ~= base, "different client model")
assert(fp(inst, other_temp) ~= base, "different temperature")
- assert(fp(inst, with_stream) == base, "the stream flag is
stripped, so it must not matter")
+ assert(fp(inst, with_stream) ~= base, "the stream flag is
folded into the fingerprint (streaming and non-streaming cache separately)")
-- instance config that ai-proxy applies upstream: a change flips
it too
local other_provider = { provider = "deepseek", options = {},
override = {} }
@@ -687,7 +687,7 @@ X-AI-Cache-Status: MISS
--- response_body_like eval
qr/1 \+ 1 = 2/
--- error_log
-ai-cache: redis unavailable, fail-open as MISS
+ai-cache: L1 lookup failed, fail-open as MISS
