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

fix

Browse Source
This commit is contained in:
Mikan
2026-06-19 17:32:21 +03:00
parent 81bbf8aa69
commit 2167493887
6 changed files with 484 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135
backend/app/api/admin.py
View File

@@ -211,6 +211,141 @@ async def test_llm_endpoint(
return result
@router.post("/llm/test-tools")
async def test_llm_tools_endpoint(
payload: Dict[str, Any] = Body(default={}),
db: AsyncSession = Depends(get_db_dep),
_: User = Depends(require_admin),
):
"""Probe whether the configured LLM endpoint supports OpenAI-style tool calls.
Sends a minimal chat completion request WITH a `tools` array containing one
simple function (`get_time`). Returns:
- ok: bool — did the model produce ANY well-formed response?
- tool_calls_returned: bool — did the model emit at least one tool_call?
- tool_call_name: str|null — the function name the model called (if any)
- tool_call_args: dict|null — the parsed arguments (if any)
- text: str — the model's text response (if any)
- http_status: int — HTTP status of the chat-completions call
- latency_ms: int
- raw_tool_calls: list — the raw tool_calls array from the response
- error: str|null — error message if the request failed
- error_type: str|null
Use this to verify the model actually supports function-calling before
relying on it for world-builder / orchestrator / step-writer flows.
"""
import time
import httpx
import json as _json
settings_map = await get_all_settings(db)
overrides = (payload or {}).get("overrides") or {}
for k, v in overrides.items():
if k in EDITABLE_SETTING_KEYS:
settings_map[k] = v
base_url = str(settings_map.get("llm.base_url", "")).rstrip("/")
model = str(settings_map.get("llm.model", "local-model"))
api_key = str(settings_map.get("llm.api_key", "dummy"))
timeout_s = float(settings_map.get("llm.request_timeout", 30) or 30)
result: Dict[str, Any] = {
"base_url": base_url,
"model": model,
"ok": False,
"tool_calls_returned": False,
}
headers = {"Content-Type": "application/json"}
if api_key and api_key != "dummy":
headers["Authorization"] = f"Bearer {api_key}"
# Minimal tool definition — the model should call this.
tools = [
{
"type": "function",
"function": {
"name": "get_current_time",
"description": "Returns the current time. Call this when the user asks for the time.",
"parameters": {
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "Optional timezone, e.g. 'UTC' or 'Europe/Moscow'",
},
},
"required": [],
},
},
}
]
chat_url = f"{base_url}/chat/completions"
chat_payload = {
"model": model,
"messages": [
{"role": "system", "content": "You are a helpful assistant. When the user asks for the time, you MUST call the get_current_time tool."},
{"role": "user", "content": "What time is it now? Use the get_current_time tool to find out."},
],
"tools": tools,
"tool_choice": "auto",
"max_tokens": 200,
"temperature": 0.0,
"stream": False,
}
started = time.monotonic()
try:
async with httpx.AsyncClient(timeout=httpx.Timeout(connect=10.0, read=timeout_s, write=10.0, pool=5.0)) as client:
resp = await client.post(chat_url, json=chat_payload, headers=headers)
result["http_status"] = resp.status_code
result["latency_ms"] = int((time.monotonic() - started) * 1000)
if resp.status_code >= 400:
result["error"] = f"HTTP {resp.status_code}: {resp.text[:500]}"
result["error_type"] = "HTTPError"
return result
data = resp.json()
except httpx.ConnectError as e:
cause = getattr(e, "__cause__", None) or getattr(e, "__context__", None)
result["error"] = f"ConnectError: {e}"
if cause:
result["error"] += f" (cause: {cause})"
result["error_type"] = "ConnectError"
return result
except Exception as e:
result["error"] = f"{type(e).__name__}: {e}"
result["error_type"] = type(e).__name__
return result
try:
choice = (data.get("choices") or [{}])[0]
msg = choice.get("message", {})
text = msg.get("content") or ""
tool_calls = msg.get("tool_calls") or []
result["text"] = text[:500]
result["raw_tool_calls"] = tool_calls
if tool_calls:
result["tool_calls_returned"] = True
first = tool_calls[0]
fn = first.get("function", {}) if isinstance(first, dict) else {}
result["tool_call_name"] = fn.get("name")
args_str = fn.get("arguments", "{}")
try:
result["tool_call_args"] = _json.loads(args_str) if args_str else {}
except _json.JSONDecodeError:
result["tool_call_args"] = {"_raw": args_str}
result["ok"] = True
result["usage"] = data.get("usage", {})
except Exception as e:
result["error"] = f"response_parse_failed: {type(e).__name__}: {e}"
result["error_type"] = type(e).__name__
return result
return result
@router.get("/llm-logs", response_model=List[LlmLogOut])
async def list_llm_logs(
limit: int = 50,

210
backend/app/engine/world_builder.py
View File

@@ -83,7 +83,9 @@ async def start_world_builder(
db=db,
)
ai_text, proposed, is_final = _extract_world_definition(response.text, response.tool_calls)
ai_text, proposed, is_final = _extract_world_definition(
response.text, response.tool_calls, user_confirmation=False,
)
_DIALOGUES[session_id] = {
"user_id": user.id,
@@ -129,6 +131,11 @@ async def continue_world_builder(
dialogue["messages"].append({"role": "user", "content": user_message})
dialogue["turn"] += 1
# Detect if the player is explicitly confirming the world is ready.
# If so, we'll force is_final=True on the extracted definition (the model
# often forgets to set is_final even when the player clearly accepted).
user_confirms = _is_user_confirmation(user_message)
response = await llm.chat(
messages=dialogue["messages"],
tools=WORLD_BUILDER_TOOL_SCHEMAS,
@@ -143,7 +150,17 @@ async def continue_world_builder(
"tool_calls": response.tool_calls or None,
})
ai_text, proposed, is_final = _extract_world_definition(response.text, response.tool_calls)
ai_text, proposed, is_final = _extract_world_definition(
response.text, response.tool_calls, user_confirmation=user_confirms,
)
# If the model didn't propose a new definition this turn but we already had
# one stored and the player just confirmed, reuse the stored definition
# and mark it final.
if proposed is None and user_confirms and dialogue.get("last_proposed"):
proposed = WorldDefinition.model_validate(dialogue["last_proposed"])
is_final = True
if proposed:
dialogue["last_proposed"] = proposed.model_dump()
@@ -220,74 +237,205 @@ def _build_user_brief(
return "\n".join(parts)
# Phrases in EN/RU that the player might type to confirm a proposed world is
# ready to commit. Matched case-insensitively against the player's message.
# Keep this list SHORT and specific — false positives would auto-finalize a
# world the player didn't intend to accept.
_CONFIRMATION_PHRASES = (
# English
"ready", "looks good", "looks great", "perfect", "ok", "okay", "fine",
"yes", "yep", "yeah", "sure", "go ahead", "confirm", "confirmed",
"approve", "approved", "let's go", "lets go", "do it", "lgtm",
"i'm happy", "im happy", "ship it", "all good",
# Russian
"готово", "готов", "супер", "ок", "окей", "хорошо", "отлично",
"да", "согласен", "согласна", "подтверждаю", "одобряю", "норм",
"нормально", "поехали", "создай", "сохраняй", "принимаю",
)
def _is_user_confirmation(message: str) -> bool:
"""Return True if the player's message looks like explicit confirmation.
Heuristic: the message is short (under 60 chars) AND contains one of the
known confirmation phrases. Longer messages are treated as edits/feedback,
not confirmation, even if they contain a "yes".
"""
if not message:
return False
msg = message.strip().lower()
if not msg or len(msg) > 60:
return False
# Exact match or substring — both work. The phrase list is short enough
# that false positives are rare in normal player edits.
return any(phrase in msg for phrase in _CONFIRMATION_PHRASES)
def _extract_world_definition(
text: str,
tool_calls: Optional[List[Dict[str, Any]]],
user_confirmation: bool = False,
) -> tuple[str, Optional[WorldDefinition], bool]:
"""Extract AI message text, proposed definition (if any), and is_final flag.
Looks for a `submit_world_definition` tool call first. Falls back to
JSON-in-text parse for older models that don't honor the tool.
Strategy (in order):
1. If the model returned a `submit_world_definition` tool_call — use its
arguments directly. This is the preferred path for models that support
OpenAI-style function calling.
2. Otherwise, scan the text for a JSON object (in a ```json fenced block
or a bare `{...}` block). Many smaller local models emit the structured
payload as text instead of using tool_calls; we still want to honor it.
3. If `user_confirmation=True` (the player just said something like
"ok" / "ready" / "go") and we have a proposed definition, force
`is_final=True` even if the model forgot to set it.
"""
proposed: Optional[WorldDefinition] = None
is_final = False
ai_text = text or ""
ai_text = _sanitize_model_text(text or "")
# 1) Prefer the submit_world_definition tool call (the proper way).
if tool_calls:
for tc in tool_calls:
if tc.get("function", {}).get("name") == "submit_world_definition":
args_str = tc.get("function", {}).get("arguments", "{}")
try:
data = json.loads(args_str) if args_str else {}
except json.JSONDecodeError:
data = {}
proposed = _try_build_definition(data)
is_final = bool(data.get("is_final", False))
break
data = _safe_json_loads(args_str, {})
if data:
proposed = _try_build_definition(data)
is_final = bool(data.get("is_final", False))
break
# 2) Fallback: parse a JSON block from the text (older models).
# 2) Fallback: parse a JSON block from the text (models without tool_calls).
if proposed is None:
json_str = _extract_json_block(text)
json_str = _extract_json_block(ai_text)
if json_str:
try:
data = json.loads(json_str)
if isinstance(data, dict):
if "proposed_definition" in data and isinstance(data["proposed_definition"], dict):
proposed = _try_build_definition(data["proposed_definition"])
data = _safe_json_loads(json_str, None)
if isinstance(data, dict):
# Accept either a flat world-definition object or a wrapper
# like {"proposed_definition": {...}, "is_final": bool, "ai_message": "..."}.
target = data
if "proposed_definition" in data and isinstance(data["proposed_definition"], dict):
target = data["proposed_definition"]
if "is_final" in data:
is_final = bool(data["is_final"])
if "ai_message" in data and isinstance(data["ai_message"], str):
# Strip the JSON wrapper from the visible text so the
# player doesn't see raw JSON in chat.
ai_text = data["ai_message"]
except json.JSONDecodeError:
pass
else:
if "is_final" in data:
is_final = bool(data["is_final"])
proposed = _try_build_definition(target)
# 3) Heuristic: if a definition was proposed and the text mentions "ready",
# mark as final.
if proposed is not None and not is_final:
low = ai_text.lower()
if any(kw in low for kw in ["world is ready", "world_ready", "ready to commit", "мир готов"]):
is_final = True
# 3) If the player just confirmed and we have a definition, force is_final.
if proposed is not None and user_confirmation and not is_final:
is_final = True
# 4) Strip any leaked JSON block from the visible AI text so the player
# never sees raw JSON in chat. Keep the prose portion only.
if proposed is not None:
ai_text = _strip_json_blocks(ai_text).strip()
if not ai_text:
# Model returned only JSON with no prose — synthesize a short
# confirmation message in the player's language.
ai_text = "(definition ready)"
return ai_text, proposed, is_final
def _sanitize_model_text(text: str) -> str:
"""Remove common model-output artifacts that would break JSON parsing.
Strips:
- End-of-sequence tokens (model-specific control tokens that occasionally
leak into the decoded text).
- Leading/trailing whitespace per line.
- Empty leading lines.
"""
if not text:
return ""
# Remove EOS-style control tokens (single token on its own line, or
# repeated tokens, or trailing tokens).
import re as _re
# Collapse repeated EOS tokens anywhere in the text.
cleaned = _re.sub(r"<\s*/?\s*[a-zA-Z]+\s*>", "", text)
# Collapse 3+ blank lines into 1.
cleaned = _re.sub(r"\n{3,}", "\n\n", cleaned)
return cleaned.strip()
def _strip_json_blocks(text: str) -> str:
"""Remove fenced ```json ... ``` blocks and bare trailing {...} blocks
from `text`. Used to clean the player-visible AI message after we've
already extracted the structured data.
"""
if not text:
return ""
import re as _re
# Strip fenced code blocks (any language).
cleaned = _re.sub(r"```[a-zA-Z]*\s*[\s\S]*?```", "", text)
# Strip trailing bare JSON object (last {...} block in the text).
# We only remove it if it's the LAST substantial thing in the text,
# to avoid mangling prose that legitimately contains braces.
m = _re.search(r"\n\{[\s\S]*\}\s*$", cleaned)
if m:
cleaned = cleaned[: m.start()] + cleaned[m.end():]
return cleaned
def _safe_json_loads(s: str, default: Any) -> Any:
"""Tolerant JSON loader. Returns `default` on failure.
Attempts standard json.loads first. If that fails, tries to repair common
model-output mistakes:
- Trailing commas before } or ].
- Single quotes instead of double quotes.
- Unescaped newlines inside string values.
"""
if not s:
return default
try:
return json.loads(s)
except json.JSONDecodeError:
pass
# Repair attempt 1: remove trailing commas.
import re as _re
repaired = _re.sub(r",\s*([}\]])", r"\1", s)
try:
return json.loads(repaired)
except json.JSONDecodeError:
pass
# Repair attempt 2: replace single quotes with double quotes (naive, but
# catches many cases where the model emits pseudo-JSON).
try:
repaired2 = repaired.replace("'", '"')
return json.loads(repaired2)
except json.JSONDecodeError:
return default
def _try_build_definition(data: Dict[str, Any]) -> Optional[WorldDefinition]:
try:
return WorldDefinition.model_validate(data)
except Exception as e:
log.warning("world_definition_invalid", error=str(e))
log.warning("world_definition_invalid", error=str(e), keys=list(data.keys()))
return None
def _extract_json_block(text: str) -> Optional[str]:
"""Find the first JSON object/array block in text (fallback path only)."""
"""Find the first JSON object block in text.
Prefers a fenced ```json ... ``` block. Falls back to the largest balanced
{...} block in the text.
"""
if not text:
return None
import re
m = re.search(r"```(?:json)?\s*(\{[\s\S]*?\})\s*```", text)
import re as _re
# Fenced block (```json ... ``` or ``` ... ```).
m = _re.search(r"```(?:json)?\s*(\{[\s\S]*?\})\s*```", text)
if m:
return m.group(1)
# Bare block: find the first `{` and balance braces, respecting strings
# and escape sequences.
start = text.find("{")
if start == -1:
return None