fix: improvements, fixes and opt-in cache
doc: semantic-cache.md added with detailed write-up
This commit is contained in:
parent
a5108486e3
commit
fbdc73eebb
4 changed files with 598 additions and 18 deletions
85
router.py
85
router.py
|
|
@ -1600,7 +1600,8 @@ async def proxy(request: Request):
|
|||
images = payload.get("images")
|
||||
options = payload.get("options")
|
||||
keep_alive = payload.get("keep_alive")
|
||||
|
||||
_cache_enabled = payload.get("nomyo", {}).get("cache", False)
|
||||
|
||||
if not model:
|
||||
raise HTTPException(
|
||||
status_code=400, detail="Missing required field 'model'"
|
||||
|
|
@ -1615,7 +1616,7 @@ async def proxy(request: Request):
|
|||
|
||||
# Cache lookup — before endpoint selection so no slot is wasted on a hit
|
||||
_cache = get_llm_cache()
|
||||
if _cache is not None:
|
||||
if _cache is not None and _cache_enabled:
|
||||
_cached = await _cache.get_generate(model, prompt, system or "")
|
||||
if _cached is not None:
|
||||
async def _serve_cached_generate():
|
||||
|
|
@ -1671,7 +1672,7 @@ async def proxy(request: Request):
|
|||
else:
|
||||
json_line = orjson.dumps(chunk)
|
||||
# Accumulate and store cache on done chunk — before yield so it always runs
|
||||
if _cache is not None:
|
||||
if _cache is not None and _cache_enabled:
|
||||
if getattr(chunk, "response", None):
|
||||
content_parts.append(chunk.response)
|
||||
if getattr(chunk, "done", False):
|
||||
|
|
@ -1710,7 +1711,7 @@ async def proxy(request: Request):
|
|||
cache_bytes = json_line.encode("utf-8") + b"\n"
|
||||
yield cache_bytes
|
||||
# Cache non-streaming response
|
||||
if _cache is not None:
|
||||
if _cache is not None and _cache_enabled:
|
||||
try:
|
||||
await _cache.set_generate(model, prompt, system or "", cache_bytes)
|
||||
except Exception as _ce:
|
||||
|
|
@ -1749,6 +1750,7 @@ async def chat_proxy(request: Request):
|
|||
options = payload.get("options")
|
||||
logprobs = payload.get("logprobs")
|
||||
top_logprobs = payload.get("top_logprobs")
|
||||
_cache_enabled = payload.get("nomyo", {}).get("cache", False)
|
||||
|
||||
if not model:
|
||||
raise HTTPException(
|
||||
|
|
@ -1775,7 +1777,7 @@ async def chat_proxy(request: Request):
|
|||
# Snapshot original messages before any OpenAI-format transformation so that
|
||||
# get_chat and set_chat always use the same key regardless of backend type.
|
||||
_cache_messages = messages
|
||||
if _cache is not None and not _is_moe:
|
||||
if _cache is not None and not _is_moe and _cache_enabled:
|
||||
_cached = await _cache.get_chat("ollama_chat", _cache_model, messages)
|
||||
if _cached is not None:
|
||||
async def _serve_cached_chat():
|
||||
|
|
@ -1858,7 +1860,7 @@ async def chat_proxy(request: Request):
|
|||
# Accumulate and store cache on done chunk — before yield so it always runs
|
||||
# Works for both Ollama-native and OpenAI-compatible backends; chunks are
|
||||
# already converted to Ollama format by rechunk before this point.
|
||||
if _cache is not None and not _is_moe:
|
||||
if _cache is not None and not _is_moe and _cache_enabled:
|
||||
if chunk.message and getattr(chunk.message, "content", None):
|
||||
content_parts.append(chunk.message.content)
|
||||
if getattr(chunk, "done", False):
|
||||
|
|
@ -1898,7 +1900,7 @@ async def chat_proxy(request: Request):
|
|||
cache_bytes = json_line.encode("utf-8") + b"\n"
|
||||
yield cache_bytes
|
||||
# Cache non-streaming response (non-MOE; works for both Ollama and OpenAI backends)
|
||||
if _cache is not None and not _is_moe:
|
||||
if _cache is not None and not _is_moe and _cache_enabled:
|
||||
try:
|
||||
await _cache.set_chat("ollama_chat", _cache_model, _cache_messages, cache_bytes)
|
||||
except Exception as _ce:
|
||||
|
|
@ -2746,6 +2748,7 @@ async def openai_chat_completions_proxy(request: Request):
|
|||
tools = payload.get("tools")
|
||||
logprobs = payload.get("logprobs")
|
||||
top_logprobs = payload.get("top_logprobs")
|
||||
_cache_enabled = payload.get("nomyo", {}).get("cache", False)
|
||||
|
||||
if not model:
|
||||
raise HTTPException(
|
||||
|
|
@ -2786,9 +2789,20 @@ async def openai_chat_completions_proxy(request: Request):
|
|||
except orjson.JSONDecodeError as e:
|
||||
raise HTTPException(status_code=400, detail=f"Invalid JSON: {e}") from e
|
||||
|
||||
# Reject unsupported image formats (SVG) before doing any work
|
||||
for _msg in messages:
|
||||
for _item in (_msg.get("content") or []) if isinstance(_msg.get("content"), list) else []:
|
||||
if _item.get("type") == "image_url":
|
||||
_url = (_item.get("image_url") or {}).get("url", "")
|
||||
if _url.startswith("data:image/svg") or _url.lower().endswith(".svg"):
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="SVG images are not supported. Please convert the image to PNG or JPEG before sending.",
|
||||
)
|
||||
|
||||
# Cache lookup — before endpoint selection
|
||||
_cache = get_llm_cache()
|
||||
if _cache is not None:
|
||||
if _cache is not None and _cache_enabled:
|
||||
_cached = await _cache.get_chat("openai_chat", model, messages)
|
||||
if _cached is not None:
|
||||
if stream:
|
||||
|
|
@ -2806,16 +2820,56 @@ async def openai_chat_completions_proxy(request: Request):
|
|||
base_url = ep2base(endpoint)
|
||||
oclient = openai.AsyncOpenAI(base_url=base_url, default_headers=default_headers, api_key=config.api_keys.get(endpoint, "no-key"))
|
||||
# 3. Async generator that streams completions data and decrements the counter
|
||||
async def _normalize_images_in_messages(msgs: list) -> list:
|
||||
"""Fetch remote image URLs and convert them to base64 data URLs so
|
||||
Ollama/llama-server can handle them without making outbound HTTP requests."""
|
||||
resolved = []
|
||||
for msg in msgs:
|
||||
content = msg.get("content")
|
||||
if not isinstance(content, list):
|
||||
resolved.append(msg)
|
||||
continue
|
||||
new_content = []
|
||||
for item in content:
|
||||
if item.get("type") == "image_url":
|
||||
url = (item.get("image_url") or {}).get("url", "")
|
||||
if url and not url.startswith("data:"):
|
||||
try:
|
||||
http: aiohttp.ClientSession = app_state["session"]
|
||||
async with http.get(url) as resp:
|
||||
ctype = resp.headers.get("Content-Type", "image/jpeg").split(";")[0].strip()
|
||||
img_bytes = await resp.read()
|
||||
b64 = base64.b64encode(img_bytes).decode("utf-8")
|
||||
new_content.append({
|
||||
"type": "image_url",
|
||||
"image_url": {"url": f"data:{ctype};base64,{b64}"}
|
||||
})
|
||||
except Exception as _ie:
|
||||
print(f"[image] Failed to fetch image URL: {_ie}")
|
||||
new_content.append(item)
|
||||
else:
|
||||
new_content.append(item)
|
||||
else:
|
||||
new_content.append(item)
|
||||
resolved.append({**msg, "content": new_content})
|
||||
return resolved
|
||||
|
||||
async def stream_ochat_response():
|
||||
try:
|
||||
# The chat method returns a generator of dicts (or GenerateResponse)
|
||||
try:
|
||||
async_gen = await oclient.chat.completions.create(**params)
|
||||
# For non-external endpoints (Ollama, llama-server), resolve remote
|
||||
# image URLs to base64 data URLs so the server can handle them locally.
|
||||
send_params = params
|
||||
if not is_ext_openai_endpoint(endpoint):
|
||||
resolved_msgs = await _normalize_images_in_messages(params.get("messages", []))
|
||||
send_params = {**params, "messages": resolved_msgs}
|
||||
async_gen = await oclient.chat.completions.create(**send_params)
|
||||
except openai.BadRequestError as e:
|
||||
# If tools are not supported by the model, retry without tools
|
||||
if "does not support tools" in str(e):
|
||||
print(f"[openai_chat_completions_proxy] Model {model} doesn't support tools, retrying without tools")
|
||||
params_without_tools = {k: v for k, v in params.items() if k != "tools"}
|
||||
params_without_tools = {k: v for k, v in send_params.items() if k != "tools"}
|
||||
async_gen = await oclient.chat.completions.create(**params_without_tools)
|
||||
else:
|
||||
raise
|
||||
|
|
@ -2856,7 +2910,7 @@ async def openai_chat_completions_proxy(request: Request):
|
|||
if prompt_tok != 0 or comp_tok != 0:
|
||||
await token_queue.put((endpoint, tracking_model, prompt_tok, comp_tok))
|
||||
# Cache assembled streaming response — before [DONE] so it always runs
|
||||
if _cache is not None and content_parts:
|
||||
if _cache is not None and _cache_enabled and content_parts:
|
||||
assembled = orjson.dumps({
|
||||
"model": model,
|
||||
"choices": [{"index": 0, "message": {"role": "assistant", "content": "".join(content_parts)}, "finish_reason": "stop"}],
|
||||
|
|
@ -2887,7 +2941,7 @@ async def openai_chat_completions_proxy(request: Request):
|
|||
cache_bytes = json_line.encode("utf-8") + b"\n"
|
||||
yield cache_bytes
|
||||
# Cache non-streaming response
|
||||
if _cache is not None:
|
||||
if _cache is not None and _cache_enabled:
|
||||
try:
|
||||
await _cache.set_chat("openai_chat", model, messages, cache_bytes)
|
||||
except Exception as _ce:
|
||||
|
|
@ -2930,6 +2984,7 @@ async def openai_completions_proxy(request: Request):
|
|||
max_tokens = payload.get("max_tokens")
|
||||
max_completion_tokens = payload.get("max_completion_tokens")
|
||||
suffix = payload.get("suffix")
|
||||
_cache_enabled = payload.get("nomyo", {}).get("cache", False)
|
||||
|
||||
if not model:
|
||||
raise HTTPException(
|
||||
|
|
@ -2970,7 +3025,7 @@ async def openai_completions_proxy(request: Request):
|
|||
# Cache lookup — completions prompt mapped to a single-turn messages list
|
||||
_cache = get_llm_cache()
|
||||
_compl_messages = [{"role": "user", "content": prompt}]
|
||||
if _cache is not None:
|
||||
if _cache is not None and _cache_enabled:
|
||||
_cached = await _cache.get_chat("openai_completions", model, _compl_messages)
|
||||
if _cached is not None:
|
||||
if stream:
|
||||
|
|
@ -3029,7 +3084,7 @@ async def openai_completions_proxy(request: Request):
|
|||
if prompt_tok != 0 or comp_tok != 0:
|
||||
await token_queue.put((endpoint, tracking_model, prompt_tok, comp_tok))
|
||||
# Cache assembled streaming response — before [DONE] so it always runs
|
||||
if _cache is not None and text_parts:
|
||||
if _cache is not None and _cache_enabled and text_parts:
|
||||
assembled = orjson.dumps({
|
||||
"model": model,
|
||||
"choices": [{"index": 0, "message": {"role": "assistant", "content": "".join(text_parts)}, "finish_reason": "stop"}],
|
||||
|
|
@ -3061,7 +3116,7 @@ async def openai_completions_proxy(request: Request):
|
|||
cache_bytes = json_line.encode("utf-8") + b"\n"
|
||||
yield cache_bytes
|
||||
# Cache non-streaming response
|
||||
if _cache is not None:
|
||||
if _cache is not None and _cache_enabled:
|
||||
try:
|
||||
await _cache.set_chat("openai_completions", model, _compl_messages, cache_bytes)
|
||||
except Exception as _ce:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue