262 lines
11 KiB
Python
262 lines
11 KiB
Python
"""Elements (personal summary) and tutor chat for the guide."""
|
|
|
|
import json
|
|
import logging
|
|
import uuid
|
|
|
|
from agents import run_agent
|
|
from config import DEFAULT_PROVIDER
|
|
from jsonio import parse_json_text as _parse_json_text, read_json_file as _read_json_file
|
|
from paths import blocks_path, guide_content_path
|
|
from pipeline import _prompt
|
|
|
|
log = logging.getLogger("creator.elements")
|
|
|
|
|
|
# --- Tutor chat ---
|
|
|
|
def _build_guide_chat_prompt(topic: str, format_name: str, section: str, outline: str, messages: list[dict]) -> str:
|
|
transcript = "\n".join(
|
|
f"{'User' if m.get('role') == 'user' else 'Assistant'}: {m.get('content', '')}"
|
|
for m in messages
|
|
)
|
|
return _prompt(
|
|
"Chat",
|
|
topic=topic, format_name=format_name,
|
|
outline_block=outline.strip() or "(none)",
|
|
section_block=section.strip() or "(no section detected)",
|
|
transcript=transcript,
|
|
)
|
|
|
|
|
|
async def chat_with_guide(topic: str, format_name: str, section: str, outline: str, messages: list[dict], provider: str = DEFAULT_PROVIDER) -> str:
|
|
try:
|
|
prompt = _build_guide_chat_prompt(topic, format_name, section, outline, messages)
|
|
returncode, stdout, stderr = await run_agent(
|
|
"chat-" + str(uuid.uuid4()), prompt, 240, provider=provider, role="fast", capabilities="none", lane="interactive"
|
|
)
|
|
if returncode != 0:
|
|
return "Sorry, that didn't work. Please try again."
|
|
reply = stdout.strip()
|
|
return reply or "Sorry, I didn't get a response."
|
|
except Exception:
|
|
log.warning("[%s] Guide chat failed", topic, exc_info=True)
|
|
return "Sorry, that didn't work. Please try again."
|
|
|
|
|
|
# --- Elements ---
|
|
|
|
def _element_fields(data: dict) -> dict | None:
|
|
"""Validate AI element JSON and normalize it onto the DB fields."""
|
|
if not isinstance(data, dict):
|
|
return None
|
|
title = str(data.get("title", "")).strip()
|
|
if not title:
|
|
return None
|
|
lists = {}
|
|
for key in ("examples", "hints"):
|
|
raw = data.get(key, [])
|
|
lists[key] = [str(e).strip() for e in raw if str(e).strip()] if isinstance(raw, list) else []
|
|
return {
|
|
"title": title[:200],
|
|
"description": str(data.get("description", "")).strip(),
|
|
"examples": lists["examples"],
|
|
"hints": lists["hints"],
|
|
}
|
|
|
|
|
|
def _topic_context(topic: str, limit: int = 12000) -> str:
|
|
"""Blocks + guide content of the topic as context text (truncated)."""
|
|
parts: list[str] = []
|
|
bp = blocks_path(topic)
|
|
if bp.exists():
|
|
parts.append(bp.read_text(encoding="utf-8"))
|
|
for fmt in ("Guide", "FullGuide"): # best available prose guide as chat context
|
|
content = _read_json_file(guide_content_path(topic, fmt))
|
|
if content:
|
|
for ch in content.get("chapters", []):
|
|
for sec in ch.get("sections", []):
|
|
parts.append(sec if isinstance(sec, str) else json.dumps(sec, ensure_ascii=False))
|
|
break # the best available guide is enough
|
|
text = "\n\n".join(parts).strip()
|
|
return text[:limit] if text else "(no material available)"
|
|
|
|
|
|
async def generate_element(topic: str, hint: str, provider: str = DEFAULT_PROVIDER, extra_context: str = "") -> dict:
|
|
"""Create element fields via AI. Fallback: only the title from the keyword."""
|
|
fallback = {"title": hint.strip() or "New element", "description": "", "examples": [], "hints": []}
|
|
try:
|
|
context = _topic_context(topic)
|
|
if extra_context.strip():
|
|
context = (extra_context.strip() + "\n\n" + context)[:12000]
|
|
prompt = _prompt(
|
|
"Element-Create",
|
|
topic=topic, hint=hint.strip() or "(none — pick a core concept yourself)",
|
|
context=context,
|
|
)
|
|
returncode, stdout, _ = await run_agent(
|
|
"element-" + str(uuid.uuid4()), prompt, 240, provider=provider, role="fast", capabilities="none", lane="interactive"
|
|
)
|
|
if returncode != 0:
|
|
return fallback
|
|
return _element_fields(_parse_json_text(stdout)) or fallback
|
|
except Exception:
|
|
log.warning("[%s] Element creation failed", topic, exc_info=True)
|
|
return fallback
|
|
|
|
|
|
def _parse_suggestions(stdout: str) -> list[dict] | None:
|
|
"""Validate suggestion JSON from AI output. None on invalid JSON."""
|
|
data = _parse_json_text(stdout)
|
|
if not isinstance(data, dict):
|
|
return None
|
|
suggestions = []
|
|
for s in data.get("suggestions", []):
|
|
if not isinstance(s, dict):
|
|
continue
|
|
text = str(s.get("text", "")).strip()
|
|
target = s.get("target")
|
|
content = str(s.get("content", "")).strip()
|
|
if text and content and target in ("description", "examples", "hints"):
|
|
suggestions.append({"text": text, "target": target, "content": content})
|
|
return suggestions
|
|
|
|
|
|
async def check_element(element: dict, provider: str = DEFAULT_PROVIDER) -> list[dict] | None:
|
|
"""Two-step check for missing info: research → verify. None on error."""
|
|
try:
|
|
element_json = json.dumps(
|
|
{k: element[k] for k in ("title", "description", "examples", "hints")},
|
|
ensure_ascii=False, indent=1,
|
|
)
|
|
context = _topic_context(element["topic"])
|
|
|
|
# Step 1: research — collect candidates broadly
|
|
prompt = _prompt("Element-Check", topic=element["topic"], element_json=element_json, context=context)
|
|
returncode, stdout, _ = await run_agent(
|
|
"element-check-" + str(uuid.uuid4()), prompt, 240, provider=provider, role="fast", capabilities="none", lane="interactive"
|
|
)
|
|
if returncode != 0:
|
|
return None
|
|
candidates = _parse_suggestions(stdout)
|
|
if candidates is None:
|
|
return None
|
|
if not candidates:
|
|
return []
|
|
|
|
# Step 2: verify — only let important, non-redundant items through
|
|
prompt = _prompt(
|
|
"Element-Verify",
|
|
topic=element["topic"], element_json=element_json,
|
|
candidates_json=json.dumps({"suggestions": candidates}, ensure_ascii=False, indent=1),
|
|
context=context,
|
|
)
|
|
returncode, stdout, _ = await run_agent(
|
|
"element-verify-" + str(uuid.uuid4()), prompt, 240, provider=provider, role="fast", capabilities="none", lane="interactive"
|
|
)
|
|
if returncode != 0:
|
|
return None
|
|
return _parse_suggestions(stdout)
|
|
except Exception:
|
|
log.warning("[%s] Element check failed", element.get("topic", "?"), exc_info=True)
|
|
return None
|
|
|
|
|
|
def _element_json(element: dict) -> str:
|
|
return json.dumps(
|
|
{k: element[k] for k in ("title", "description", "examples", "hints")},
|
|
ensure_ascii=False, indent=1,
|
|
)
|
|
|
|
|
|
def _validate_change(c, element: dict) -> dict | None:
|
|
"""Validate a change suggestion from AI output against the element."""
|
|
if not isinstance(c, dict):
|
|
return None
|
|
text = str(c.get("text", "")).strip()
|
|
action = c.get("action")
|
|
target = c.get("target")
|
|
index = c.get("index")
|
|
content = str(c.get("content", "")).strip()
|
|
if not text or action not in ("remove", "adjust", "add"):
|
|
return None
|
|
if target not in ("title", "description", "examples", "hints"):
|
|
return None
|
|
if action in ("adjust", "add") and not content:
|
|
return None
|
|
if action == "remove" and target not in ("examples", "hints"):
|
|
return None
|
|
# Index only for adjust/remove on list fields; must exist
|
|
if target in ("examples", "hints") and action in ("adjust", "remove"):
|
|
if not isinstance(index, int) or not (0 <= index < len(element[target])):
|
|
return None
|
|
else:
|
|
index = None
|
|
return {"text": text, "action": action, "target": target, "index": index, "content": content}
|
|
|
|
|
|
async def chat_with_element(element: dict, messages: list[dict], provider: str = DEFAULT_PROVIDER) -> tuple[str, list[dict]]:
|
|
"""Chat about the element. Returns (reply, change suggestions) — changes nothing directly."""
|
|
error = "Sorry, that didn't work. Please try again."
|
|
try:
|
|
transcript = "\n".join(
|
|
f"{'User' if m.get('role') == 'user' else 'Assistant'}: {m.get('content', '')}"
|
|
for m in messages
|
|
)
|
|
prompt = _prompt("Element-Chat", topic=element["topic"], element_json=_element_json(element), transcript=transcript)
|
|
returncode, stdout, _ = await run_agent(
|
|
"element-chat-" + str(uuid.uuid4()), prompt, 240, provider=provider, role="fast", capabilities="none", lane="interactive"
|
|
)
|
|
if returncode != 0:
|
|
return error, []
|
|
data = _parse_json_text(stdout)
|
|
if not isinstance(data, dict):
|
|
return error, []
|
|
changes = [v for c in data.get("changes", []) if (v := _validate_change(c, element))]
|
|
reply = str(data.get("reply", "")).strip() or ("Suggestions created." if changes else error)
|
|
return reply, changes
|
|
except Exception:
|
|
log.warning("[%s] Element chat failed", element.get("topic", "?"), exc_info=True)
|
|
return error, []
|
|
|
|
|
|
async def style_element(element: dict, provider: str = DEFAULT_PROVIDER) -> list[dict] | None:
|
|
"""Check an element against the style rules and suggest changes. None on error."""
|
|
try:
|
|
prompt = _prompt("Element-Style", topic=element["topic"], element_json=_element_json(element))
|
|
returncode, stdout, _ = await run_agent(
|
|
"element-stil-" + str(uuid.uuid4()), prompt, 240, provider=provider, role="fast", capabilities="none", lane="interactive"
|
|
)
|
|
if returncode != 0:
|
|
return None
|
|
data = _parse_json_text(stdout)
|
|
if not isinstance(data, dict):
|
|
return None
|
|
return [v for c in data.get("changes", []) if (v := _validate_change(c, element))]
|
|
except Exception:
|
|
log.warning("[%s] Style check failed", element.get("topic", "?"), exc_info=True)
|
|
return None
|
|
|
|
|
|
async def refine_suggestion(element: dict, suggestion: dict, instruction: str, provider: str = DEFAULT_PROVIDER) -> dict | None:
|
|
"""Revise a single suggestion per user instruction. None on error."""
|
|
try:
|
|
prompt = _prompt(
|
|
"Element-Refine",
|
|
topic=element["topic"], element_json=_element_json(element),
|
|
suggestion_json=json.dumps(suggestion, ensure_ascii=False, indent=1),
|
|
instruction=instruction,
|
|
)
|
|
returncode, stdout, _ = await run_agent(
|
|
"element-refine-" + str(uuid.uuid4()), prompt, 240, provider=provider, role="fast", capabilities="none", lane="interactive"
|
|
)
|
|
if returncode != 0:
|
|
return None
|
|
data = _parse_json_text(stdout)
|
|
if not isinstance(data, dict):
|
|
return None
|
|
return _validate_change(data.get("change"), element)
|
|
except Exception:
|
|
log.warning("[%s] Suggestion revision failed", element.get("topic", "?"), exc_info=True)
|
|
return None
|