File size: 2,735 Bytes
85d6f96 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 | """
Structured output parser for Lumi.
Qwen3 outputs in its native thinking-first format:
<think>
Internal reasoning — NEVER sent to TTS or shown to user.
</think>
[avatar_tag] Short opening line.
Full warm companion response continues here.
TTS latency trick: stream until </think> appears, then fire TTS immediately
on the opening line that follows.
parse_structured_output() returns a dict with:
- avatar_tag : one of the 5 states (defaults to "smile" if missing)
- opening_line : fires TTS after </think> is seen (latency-hiding trick)
- full_response : complete response text (think block stripped)
"""
import re
VALID_TAGS = {"smile", "nod", "concerned", "gentle", "laugh"}
_TAG_RE = re.compile(r"\[(\w+)\]") # anywhere in text
_TAG_START = re.compile(r"^\[(\w+)\]") # at start of text
_THINK_RE = re.compile(r"<think>.*?</think>", re.DOTALL)
def parse_structured_output(text: str) -> dict:
text = text.strip()
# Strip think block first — works for both tag-first and think-first formats
clean = _THINK_RE.sub("", text).strip()
# --- avatar tag: prefer at start of clean text, fall back to anywhere ----
tag_match = _TAG_START.match(clean) or _TAG_RE.search(clean)
if tag_match and tag_match.group(1).lower() in VALID_TAGS:
avatar_tag = tag_match.group(1).lower()
else:
avatar_tag = "smile"
# --- opening line: first line of the clean text after removing the tag ---
tag_bracket = f"[{avatar_tag}]"
body = clean.replace(tag_bracket, "").replace(tag_bracket.upper(), "").strip()
lines = [l.strip() for l in body.split("\n") if l.strip()]
opening_line = lines[0] if lines else body[:80]
full_response = body
return {
"avatar_tag": avatar_tag,
"opening_line": opening_line, # fire TTS after </think> token
"full_response": full_response, # append to TTS queue
}
def extract_facts_from_response(text: str) -> list[str]:
"""
Heuristically extract memorable facts from a model response.
Used to build the patient's persistent memory in ChromaDB.
"""
facts = []
lines = text.split(".")
for line in lines:
line = line.strip()
# keep lines that mention personal details, preferences, family
keywords = ("granddaughter", "grandson", "daughter", "son", "wife",
"husband", "likes", "loves", "favourite", "favorite",
"anniversary", "birthday", "garden", "music", "hymn",
"lonely", "feels", "misses", "remembers")
if any(k in line.lower() for k in keywords) and len(line) > 15:
facts.append(line)
return facts[:5] # cap at 5 per turn
|