mikan/ai-rpg
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rebase

Browse Source
This commit is contained in:
Mikan
2026-06-21 04:11:38 +03:00
parent bd85e186dc
commit 4dee4fb0a8
19 changed files with 1423 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
app/api/admin.py
View File

@@ -217,8 +217,34 @@ async def stats(
}
# --------------------------------------------------------------------------- #
# Helpers for test endpoints
# --------------------------------------------------------------------------- #
def _is_masked(value: str | None) -> bool:
"""Detect masked secret values (contain '…' or are exactly '****').
The admin GET /settings endpoint masks secret keys before sending them to
the client. If the client sends a masked value back to a test endpoint
(because it pre-filled the form from the masked settings response), we
must ignore it and fall back to the raw value from the DB.
"""
if not value:
return False
return "…" in value or value == "****"
def _resolve(value: str | None, fallback: str) -> str:
"""Use `value` if it's a non-empty, non-masked string; otherwise use fallback."""
if value and not _is_masked(value):
return value
return fallback or ""
# --------------------------------------------------------------------------- #
# Test endpoints — LLM, embeddings, embeddings probe dimension
# All test endpoints accept query params AND fall back to DB-stored settings.
# Masked values (containing '…' or '****') are ignored — they come from the
# admin UI's pre-filled form which displays masked secrets.
# --------------------------------------------------------------------------- #
@router.post("/test/llm")
async def test_llm(
@@ -229,9 +255,9 @@ async def test_llm(
_user: User = Depends(require_admin),
) -> dict:
settings = await get_all_settings(db)
api_url = api_url or settings.get("llm.api_url", "")
api_key = api_key or settings.get("llm.api_key", "")
model = model or settings.get("llm.model", "")
api_url = _resolve(api_url, settings.get("llm.api_url", ""))
api_key = _resolve(api_key, settings.get("llm.api_key", ""))
model = _resolve(model, settings.get("llm.model", ""))
if not api_url:
return {"ok": False, "error": {"code": "not_configured", "message": "llm.api_url is empty"},
"elapsed_ms": 0}
@@ -265,37 +291,49 @@ async def test_llm_tools(
_user: User = Depends(require_admin),
) -> dict:
settings = await get_all_settings(db)
api_url = api_url or settings.get("llm.api_url", "")
api_key = api_key or settings.get("llm.api_key", "")
model = model or settings.get("llm.model", "")
api_url = _resolve(api_url, settings.get("llm.api_url", ""))
api_key = _resolve(api_key, settings.get("llm.api_key", ""))
model = _resolve(model, settings.get("llm.model", ""))
if not api_url:
return {"ok": False, "error": {"code": "not_configured", "message": "llm.api_url is empty"},
"elapsed_ms": 0, "has_tool_calls": False}
client = LlmClient(api_url=api_url, api_key=api_key, model=model, timeout=15.0, max_retries=1)
client = LlmClient(api_url=api_url, api_key=api_key, model=model, timeout=30.0, max_retries=1)
start = time.monotonic()
try:
tools = [{
"type": "function",
"function": {
"name": "calc",
"description": "Evaluate a math expression",
"description": "Evaluate a math expression. You MUST call this tool.",
"parameters": {
"type": "object",
"required": ["expression"],
"properties": {"expression": {"type": "string"}},
"properties": {"expression": {"type": "string", "description": "e.g. '2+2'"}},
},
},
}]
resp = await client.complete(
stage="test_llm_tools",
messages=[{"role": "user", "content": "What is 2+2? Use the calc tool."}],
tools=tools, temperature=0.0, max_tokens=100,
messages=[
{"role": "system", "content": "You must use the calc tool to answer math questions. Do not compute in your head."},
{"role": "user", "content": "What is 2+2? You MUST call the calc tool with expression '2+2'."},
],
tools=tools,
tool_choice="auto",
temperature=0.0, max_tokens=100,
session=db,
)
elapsed = int((time.monotonic() - start) * 1000)
tcs = resp["message"].get("tool_calls") or []
# Also try to parse tool calls from content (some models emit them as text)
if not tcs:
content = resp["message"].get("content", "") or ""
parsed_tcs = _parse_text_tool_calls(content)
if parsed_tcs:
tcs = parsed_tcs
return {
"ok": True, "tool_calls": tcs, "has_tool_calls": bool(tcs), "elapsed_ms": elapsed,
"raw_response": resp["message"],
}
except Exception as e: # noqa: BLE001
elapsed = int((time.monotonic() - start) * 1000)
@@ -303,6 +341,50 @@ async def test_llm_tools(
"elapsed_ms": elapsed, "has_tool_calls": False}
# Pattern: call:tool_name{args} or <tool_call>name{args}</tool_call> or name(args)
import re as _re
_TOOL_CALL_PATTERNS = [
# call:name{json_args}
_re.compile(r"call:(\w+)\s*\{([^}]*)\}"),
# <tool_call>name{args}</tool_call>
_re.compile(r"<tool_call>\s*(\w+)\s*\{([^}]*)\}\s*</tool_call>"),
# name({"key": "value", ...})
_re.compile(r"(\w+)\s*\(\s*(\{[^}]*\})\s*\)"),
]
def _parse_text_tool_calls(content: str) -> list[dict]:
"""Parse tool calls emitted as text (some models don't use the OpenAI format).
Handles patterns like:
- call:calc{"expression": "2+2"}
- <tool_call>calc{"expression": "2+2"}</tool_call>
- calc({"expression": "2+2"})
"""
import json as _json
calls: list[dict] = []
for pattern in _TOOL_CALL_PATTERNS:
for match in pattern.finditer(content):
name = match.group(1)
args_str = match.group(2).strip()
try:
args = _json.loads(args_str)
except _json.JSONDecodeError:
# Try to fix common issues (single quotes, missing quotes on keys)
try:
fixed = args_str.replace("'", '"')
args = _json.loads(fixed)
except _json.JSONDecodeError:
args = {"_raw": args_str}
calls.append({
"id": f"parsed_{len(calls)}",
"type": "function",
"function": {"name": name, "arguments": _json.dumps(args)},
})
return calls
@router.post("/test/embeddings")
async def test_embeddings(
api_url: str | None = None,
@@ -324,9 +406,9 @@ async def test_embeddings(
"ok": True, "dimension": emb.dimension, "model": "offline_hash",
"first_5_values": vecs[0][:5] if vecs else [], "elapsed_ms": elapsed,
}
api_url = api_url or settings.get("embeddings.api_url") or settings.get("llm.api_url", "")
api_key = api_key or settings.get("embeddings.api_key") or settings.get("llm.api_key", "")
model = model or settings.get("embeddings.model", "")
api_url = _resolve(api_url, settings.get("embeddings.api_url") or settings.get("llm.api_url", ""))
api_key = _resolve(api_key, settings.get("embeddings.api_key") or settings.get("llm.api_key", ""))
model = _resolve(model, settings.get("embeddings.model", ""))
if not api_url:
return {"ok": False, "error": {"code": "not_configured", "message": "no api_url"},
"elapsed_ms": 0}
@@ -366,9 +448,9 @@ async def probe_dimension(
"dimension": int(settings.get("embeddings.dimension", 256)),
"elapsed_ms": 0,
}
api_url = api_url or settings.get("embeddings.api_url") or settings.get("llm.api_url", "")
api_key = api_key or settings.get("embeddings.api_key") or settings.get("llm.api_key", "")
model = model or settings.get("embeddings.model", "")
api_url = _resolve(api_url, settings.get("embeddings.api_url") or settings.get("llm.api_url", ""))
api_key = _resolve(api_key, settings.get("embeddings.api_key") or settings.get("llm.api_key", ""))
model = _resolve(model, settings.get("embeddings.model", ""))
emb = build_openai_embedder(
api_url=api_url, api_key=api_key, model=model,
dimension=int(settings.get("embeddings.dimension", 1536)),