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

atlasbot: fallback metric selection

Browse Source
This commit is contained in:
Brad Stein 2026-02-03 03:27:59 -03:00
parent b4cff2e359
commit 67302b052a
1 changed files with 24 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
atlasbot/engine/answerer.py
View File

@ -354,6 +354,20 @@ class AnswerEngine:
key_facts = _merge_fact_lines(metric_facts, key_facts) key_facts = _merge_fact_lines(metric_facts, key_facts)
if self._settings.debug_pipeline: if self._settings.debug_pipeline:
_debug_log("metric_facts_selected", {"facts": metric_facts}) _debug_log("metric_facts_selected", {"facts": metric_facts})
if not metric_facts:
if observer:
observer("retrieve", "fallback metric selection")
fallback_candidates = _filter_lines_by_keywords(summary_lines, keyword_tokens, max_lines=200)
if fallback_candidates:
metric_facts = await _select_fact_lines(
call_llm,
normalized,
fallback_candidates,
plan,
max_lines=max(2, plan.max_subquestions),
)
if metric_facts:
key_facts = _merge_fact_lines(metric_facts, key_facts)
if self._settings.debug_pipeline: if self._settings.debug_pipeline:
scored_preview = sorted( scored_preview = sorted(
[{"id": c["id"], "score": scored.get(c["id"], 0.0), "summary": c["summary"]} for c in chunks], [{"id": c["id"], "score": scored.get(c["id"], 0.0), "summary": c["summary"]} for c in chunks],
@ -1381,6 +1395,16 @@ def _strip_unknown_entities(reply: str, unknown_nodes: list[str], unknown_namesp
return cleaned or reply return cleaned or reply
def _filter_lines_by_keywords(lines: list[str], keywords: list[str], max_lines: int) -> list[str]:
if not lines:
return []
tokens = [kw.lower() for kw in keywords if isinstance(kw, str) and kw.strip()]
if not tokens:
return lines[:max_lines]
filtered = [line for line in lines if any(tok in line.lower() for tok in tokens)]
return (filtered or lines)[:max_lines]
def _lexicon_context(summary: dict[str, Any]) -> str: # noqa: C901 def _lexicon_context(summary: dict[str, Any]) -> str: # noqa: C901
if not isinstance(summary, dict): if not isinstance(summary, dict):
return "" return ""