bstein/atlasbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

atlasbot: guard metric answers with facts

Browse Source
This commit is contained in:
Brad Stein 2026-02-01 05:30:34 -03:00
parent 71d2829d0f
commit 9827db8a49
1 changed files with 23 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
atlasbot/engine/answerer.py
View File

@ -414,8 +414,8 @@ class AnswerEngine:
best_line = line best_line = line
break break
best_line = best_line or metric_facts[0] best_line = best_line or metric_facts[0]
reply_numbers = set(re.findall(r"\\d+(?:\\.\\d+)?", reply)) reply_numbers = set(re.findall(r"\d+(?:\.\d+)?", reply))
fact_numbers = set(re.findall(r"\\d+(?:\\.\\d+)?", " ".join(metric_facts))) fact_numbers = set(re.findall(r"\d+(?:\.\d+)?", " ".join(metric_facts)))
if not reply_numbers or (fact_numbers and not (reply_numbers & fact_numbers)): if not reply_numbers or (fact_numbers and not (reply_numbers & fact_numbers)):
reply = f"From the latest snapshot: {best_line}." reply = f"From the latest snapshot: {best_line}."
@ -447,6 +447,9 @@ class AnswerEngine:
if note: if note:
reply = f"{reply}\n\n{note}" reply = f"{reply}\n\n{note}"
if classify.get("question_type") in {"metric", "diagnostic"} and metric_facts:
reply = _metric_fact_guard(reply, metric_facts, keyword_tokens)
scores = await self._score_answer(normalized, reply, plan, call_llm) scores = await self._score_answer(normalized, reply, plan, call_llm)
claims = await self._extract_claims(normalized, reply, summary, call_llm) claims = await self._extract_claims(normalized, reply, summary, call_llm)
except LLMLimitReached: except LLMLimitReached:
@ -898,6 +901,24 @@ def _key_fact_lines(lines: list[str], keywords: list[str] | None, limit: int = 6
return matches return matches
def _metric_fact_guard(reply: str, metric_facts: list[str], keywords: list[str]) -> str:
if not metric_facts:
return reply
best_line = None
lowered_keywords = [kw.lower() for kw in keywords if kw]
for line in metric_facts:
line_lower = line.lower()
if any(kw in line_lower for kw in lowered_keywords):
best_line = line
break
best_line = best_line or metric_facts[0]
reply_numbers = set(re.findall(r"\d+(?:\.\d+)?", reply))
fact_numbers = set(re.findall(r"\d+(?:\.\d+)?", " ".join(metric_facts)))
if not reply_numbers or (fact_numbers and not (reply_numbers & fact_numbers)):
return f"From the latest snapshot: {best_line}."
return reply
def _lexicon_context(summary: dict[str, Any]) -> str: def _lexicon_context(summary: dict[str, Any]) -> str:
if not isinstance(summary, dict): if not isinstance(summary, dict):
return "" return ""