| """Responder: ASR text -> mix estimate -> LLM candidates -> validator (regenerate up to N rounds) -> tagged reply.""" |
| import time, re |
| import mix as M, validator as V |
| from llm import LLM, build_messages |
|
|
| class Responder: |
| def __init__(self, llm=None, n=3, rounds=2, lid="glotlid", prior_w=0.3, asr=True): |
| self.llm = llm or LLM(); self.n, self.rounds, self.lid, self.prior_w = n, rounds, lid, prior_w |
| self.asr = asr; self.style = None |
|
|
| def reply(self, user_text, history=()): |
| t0 = time.perf_counter() |
| est = M.estimate(user_text, lid=self.lid, prior=self.style, prior_w=self.prior_w if self.style else 0.0, asr=self.asr) |
| cur = M.estimate(user_text, lid=self.lid, asr=self.asr) |
| self.style = cur["mix"] if self.style is None else {k: 0.5 * self.style.get(k, 0) + 0.5 * cur["mix"].get(k, 0) for k in set(self.style) | set(cur["mix"])} |
| target = dict(mix=est["mix"], matrix=est["matrix"], pattern=cur["pattern"]) |
| desc = M.describe(dict(target, mix=est["mix"])) |
| t_mix = time.perf_counter() - t0 |
| cands, log, t_llm = [], [], 0.0 |
| chosen, how = None, None |
| for rd in range(self.rounds): |
| msgs = build_messages(user_text, desc + ("" if rd == 0 else " IMPORTANT: your previous attempt did not follow the language target or the tag format. Follow it exactly."), history) |
| outs, dt = self.llm.generate(msgs, n=self.n, temperature=0.7 if rd == 0 else 0.9) |
| t_llm += dt |
| rs = [V.check(o, target, lid=self.lid, user=user_text) for o in outs] |
| cands += rs; log.append(dict(round=rd, llm_s=round(dt, 3), cands=[dict(text=r["text"], ok=r["ok"], why=r["why"], tvd=r["tvd"]) for r in rs])) |
| ok = [r for r in rs if r["ok"]] |
| if ok: |
| chosen = min(ok, key=lambda r: r["score"]); how = "first_round" if rd == 0 else f"regen_round{rd}" |
| break |
| if chosen is None: |
| text, how = V.fallback(cands, target); chosen = V.check(text, target, lid=self.lid); chosen["text"] = text |
| return dict(user=user_text, target=target, target_desc=desc, reply=chosen["text"], reply_mix=chosen.get("mix", {}), |
| tvd=chosen.get("tvd"), how=how, first_cand_ok=bool(cands and cands[0]["ok"]), any_first_round_ok=any(r["ok"] for r in cands[:self.n]), |
| rounds=log, t_mix=round(t_mix, 4), t_llm=round(t_llm, 3), t_total=round(time.perf_counter() - t0, 3)) |
|
|
