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

atlasbot: force snapshot when metric tokens present

Browse Source
This commit is contained in:
Brad Stein 2026-02-03 19:14:50 -03:00
parent c0cdcc7e64
commit 96aabf764c
1 changed files with 18 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
atlasbot/engine/answerer.py
View File

@ -166,6 +166,7 @@ class AnswerEngine:
allowed_nodes = _allowed_nodes(summary) allowed_nodes = _allowed_nodes(summary)
allowed_namespaces = _allowed_namespaces(summary) allowed_namespaces = _allowed_namespaces(summary)
summary_lines = _summary_lines(snapshot_used) summary_lines = _summary_lines(snapshot_used)
metric_tokens = _metric_key_tokens(summary_lines)
global_facts = _global_facts(summary_lines) global_facts = _global_facts(summary_lines)
kb_summary = self._kb.summary() kb_summary = self._kb.summary()
runbooks = self._kb.runbook_titles(limit=6) runbooks = self._kb.runbook_titles(limit=6)
@ -216,6 +217,8 @@ class AnswerEngine:
classify.setdefault("follow_up", False) classify.setdefault("follow_up", False)
classify.setdefault("focus_entity", "unknown") classify.setdefault("focus_entity", "unknown")
classify.setdefault("focus_metric", "unknown") classify.setdefault("focus_metric", "unknown")
if metric_tokens and keyword_tokens and any(token in metric_tokens for token in keyword_tokens):
classify["needs_snapshot"] = True
_debug_log("route_parsed", {"classify": classify, "normalized": normalized}) _debug_log("route_parsed", {"classify": classify, "normalized": normalized})
lowered_question = f"{question} {normalized}".lower() lowered_question = f"{question} {normalized}".lower()
force_metric = bool(re.search(r"\bhow many\b|\bcount\b|\btotal\b", lowered_question)) force_metric = bool(re.search(r"\bhow many\b|\bcount\b|\btotal\b", lowered_question))
@ -1296,6 +1299,21 @@ def _hotspot_evidence(summary: dict[str, Any]) -> list[str]:
return lines return lines
def _metric_key_tokens(summary_lines: list[str]) -> set[str]:
tokens: set[str] = set()
for line in summary_lines:
if not isinstance(line, str) or ":" not in line:
continue
key = line.split(":", 1)[0].strip().lower()
if not key:
continue
tokens.add(key)
for part in re.split(r"[_\\s]+", key):
if part:
tokens.add(part)
return tokens
async def _select_best_candidate( async def _select_best_candidate(
call_llm: Callable[..., Any], call_llm: Callable[..., Any],
question: str, question: str,