atlasbot: simplify metric formatting and kb chunking

atlasbot/engine/answerer.py

@@ -2094,44 +2094,65 @@ def _format_direct_metric_line(line: str) -> str:
     if not line:
         return ""
     if ":" in line:
-        key, value = line.split(":", 1)
+        formatted = _format_colon_metric(line)
+        if formatted:
+            return formatted
+    if "=" in line:
+        formatted = _format_equals_metric(line)
+        if formatted:
+            return formatted
+    return line
+
+
+def _format_colon_metric(line: str) -> str | None:
+    key, value = line.split(":", 1)
+    key = key.strip().replace("_", " ")
+    value = value.strip()
+    if not value:
+        return None
+    if key == "nodes":
+        formatted = _format_nodes_value(value)
+        if formatted:
+            return formatted
+    if key in {"nodes total", "nodes_total"}:
+        return f"Atlas has {value} total nodes."
+    return f"{key} is {value}."
+
+
+def _format_equals_metric(line: str) -> str | None:
+    pairs: list[str] = []
+    for part in line.split(","):
+        if "=" not in part:
+            continue
+        key, value = part.split("=", 1)
         key = key.strip().replace("_", " ")
         value = value.strip()
-        if value:
+        if not value:
-            if key == "nodes":
+            continue
-                parts = [p.strip() for p in value.split(",") if p.strip()]
+        if key in {"nodes total", "nodes_total"}:
-                total = None
+            return f"Atlas has {value} total nodes."
-                rest: list[str] = []
+        pairs.append(f"{key} is {value}")
-                for part in parts:
+    if not pairs:
-                    if part.startswith("total="):
+        return None
-                        total = part.split("=", 1)[1]
+    if len(pairs) == 1:
-                    else:
+        return f"{pairs[0]}."
-                        rest.append(part.replace("_", " "))
+    return "; ".join(pairs) + "."
-                if total:
+
-                    if rest:
+
-                        return f"Atlas has {total} total nodes ({'; '.join(rest)})."
+def _format_nodes_value(value: str) -> str | None:
-                    return f"Atlas has {total} total nodes."
+    parts = [p.strip() for p in value.split(",") if p.strip()]
-            if key in {"nodes total", "nodes_total"}:
+    total = None
-                return f"Atlas has {value} total nodes."
+    rest: list[str] = []
-            return f"{key} is {value}."
+    for part in parts:
-    if "=" in line:
+        if part.startswith("total="):
-        pairs: list[str] = []
+            total = part.split("=", 1)[1]
-        for part in line.split(","):
+        else:
-            if "=" not in part:
+            rest.append(part.replace("_", " "))
-                continue
+    if not total:
-            k, v = part.split("=", 1)
+        return None
-            k = k.strip().replace("_", " ")
+    if rest:
-            v = v.strip()
+        return f"Atlas has {total} total nodes ({'; '.join(rest)})."
-            if not v:
+    return f"Atlas has {total} total nodes."
-                continue
-            if k in {"nodes total", "nodes_total"}:
-                return f"Atlas has {v} total nodes."
-            pairs.append(f"{k} is {v}")
-        if pairs:
-            if len(pairs) == 1:
-                return f"{pairs[0]}."
-            return "; ".join(pairs) + "."
-    return line
 
 
 def _global_facts(lines: list[str]) -> list[str]:

atlasbot/knowledge/loader.py

@@ -78,35 +78,52 @@ class KnowledgeBase:
 
     def chunk_lines(self, *, max_files: int = 20, max_chars: int = 6000) -> list[str]:
         self.load()
-        lines: list[str] = []
         if not self._base:
+            return []
+        lines: list[str] = []
+        self._append_summary(lines)
+        self._append_catalog(lines, max_chars)
+        if not self._within_limit(lines, max_chars):
             return lines
+        self._append_runbooks(lines)
+        if not self._within_limit(lines, max_chars):
+            return lines
+        self._append_files(lines, max_files=max_files, max_chars=max_chars)
+        return lines
+
+    def _append_summary(self, lines: list[str]) -> None:
         summary = self.summary()
         if summary:
             lines.append(f"KB Summary: {summary}")
-        # Prefer curated catalog JSON if present.
+
-        if self._atlas:
+    def _append_catalog(self, lines: list[str], max_chars: int) -> None:
-            try:
+        if not self._atlas:
-                atlas_json = json.dumps(self._atlas, indent=2)
+            return
-                lines.append("KB: atlas.json")
+        if not self._within_limit(lines, max_chars):
-                lines.extend(atlas_json.splitlines())
+            return
-            except Exception:
+        try:
-                pass
+            atlas_json = json.dumps(self._atlas, indent=2)
-        if self._runbooks:
+        except Exception:
-            lines.append("KB: runbooks.json")
+            return
-            for entry in self._runbooks:
+        lines.append("KB: atlas.json")
-                if not isinstance(entry, dict):
+        lines.extend(atlas_json.splitlines())
-                    continue
+
-                title = entry.get("title")
+    def _append_runbooks(self, lines: list[str]) -> None:
-                path = entry.get("path")
+        if not self._runbooks:
-                if title and path:
+            return
-                    lines.append(f"- {title} ({path})")
+        lines.append("KB: runbooks.json")
-        # Include markdown/text sources as additional chunks.
+        for entry in self._runbooks:
-        if len(lines) >= max_chars:
+            if not isinstance(entry, dict):
-            return lines
+                continue
+            title = entry.get("title")
+            path = entry.get("path")
+            if title and path:
+                lines.append(f"- {title} ({path})")
+
+    def _append_files(self, lines: list[str], *, max_files: int, max_chars: int) -> None:
         files = sorted(self._base.rglob("*.md")) + sorted(self._base.rglob("*.txt"))
         for path in files:
-            if len(lines) >= max_chars:
+            if not self._within_limit(lines, max_chars):
                 break
             if len(lines) > max_files * 50:
                 break
@@ -118,6 +135,7 @@ class KnowledgeBase:
                 continue
             lines.append(f"KB File: {path.relative_to(self._base)}")
             lines.extend(text.splitlines())
-            if sum(len(line) for line in lines) >= max_chars:
+
-                break
+    @staticmethod
-        return lines
+    def _within_limit(lines: list[str], max_chars: int) -> bool:
+        return sum(len(line) for line in lines) < max_chars