81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
|
import json
|
||
|
|
import urllib.error
|
||
|
|
import urllib.request
|
||
|
|
from typing import Optional
|
||
|
|
|
||
|
|
from .generator import Clause, Command, random_command
|
||
|
|
from .lexicon import COLORS, LOCATIONS, OBJECTS, VERBS
|
||
|
|
|
||
|
|
_SYSTEM_PROMPT = (
|
||
|
|
"Return ONLY a JSON object — no prose, no markdown, no code fences.\n"
|
||
|
|
"Schema: {{\"clauses\":[{{\"verb\":\"<id>\",\"object\":\"<id>\","
|
||
|
|
"\"location\":\"<id>\",\"color\":\"<id or null>\"}}],\"level\":<int>}}\n"
|
||
|
|
"Allowed verb IDs: {verbs}\n"
|
||
|
|
"Allowed object IDs: {objects}\n"
|
||
|
|
"Allowed location IDs: {locations}\n"
|
||
|
|
"Allowed color IDs: {colors}"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _system_prompt() -> str:
|
||
|
|
return _SYSTEM_PROMPT.format(
|
||
|
|
verbs=", ".join(VERBS),
|
||
|
|
objects=", ".join(OBJECTS),
|
||
|
|
locations=", ".join(LOCATIONS),
|
||
|
|
colors=", ".join(COLORS),
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _validate(data: dict, level: int) -> Optional[Command]:
|
||
|
|
clauses_raw = data.get("clauses")
|
||
|
|
if not isinstance(clauses_raw, list) or not clauses_raw:
|
||
|
|
return None
|
||
|
|
clauses = []
|
||
|
|
for c in clauses_raw:
|
||
|
|
v = c.get("verb")
|
||
|
|
o = c.get("object")
|
||
|
|
loc = c.get("location")
|
||
|
|
col = c.get("color")
|
||
|
|
if v not in VERBS or o not in OBJECTS or loc not in LOCATIONS:
|
||
|
|
return None
|
||
|
|
if col is not None and col not in COLORS:
|
||
|
|
return None
|
||
|
|
clauses.append(Clause(verb_id=v, object_id=o, location_id=loc, color_id=col))
|
||
|
|
return Command(clauses=clauses, level=level)
|
||
|
|
|
||
|
|
|
||
|
|
def llm_command(level: int, endpoint: str, model: str) -> Command:
|
||
|
|
payload = json.dumps({
|
||
|
|
"model": model,
|
||
|
|
"messages": [
|
||
|
|
{"role": "system", "content": _system_prompt()},
|
||
|
|
{"role": "user", "content": f"Generate a level {level} aphasia therapy command."},
|
||
|
|
],
|
||
|
|
"temperature": 0.8,
|
||
|
|
}).encode()
|
||
|
|
|
||
|
|
try:
|
||
|
|
req = urllib.request.Request(
|
||
|
|
endpoint,
|
||
|
|
data=payload,
|
||
|
|
headers={"Content-Type": "application/json"},
|
||
|
|
method="POST",
|
||
|
|
)
|
||
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
||
|
|
body = json.loads(resp.read())
|
||
|
|
|
||
|
|
content = body["choices"][0]["message"]["content"]
|
||
|
|
start = content.find("{")
|
||
|
|
end = content.rfind("}") + 1
|
||
|
|
if start == -1 or end == 0:
|
||
|
|
raise ValueError("brak JSON w odpowiedzi")
|
||
|
|
data = json.loads(content[start:end])
|
||
|
|
|
||
|
|
cmd = _validate(data, level)
|
||
|
|
if cmd is None:
|
||
|
|
raise ValueError("nieznane identyfikatory w odpowiedzi LLM")
|
||
|
|
return cmd
|
||
|
|
|
||
|
|
except Exception:
|
||
|
|
return random_command(level)
|