"""Read Agency Sphinx singlehtml help (TOC + semantic section search)."""
from __future__ import annotations
from pathlib import Path
from bs4 import BeautifulSoup, NavigableString, Tag
from django.conf import settings
from langchain.tools import tool
_HEADING_TAGS = frozenset({"h1", "h2", "h3", "h4", "h5", "h6"})
_STRIP_TAGS = ("script", "style", "nav", "header", "footer", "aside")
_BLOCK_TAGS = frozenset({"p", "li", "pre", "td", "th", "dt", "dd", "blockquote"})
[docs]
def singlehtml_path() -> Path:
root = Path(getattr(settings, "STATIC_ROOT", "") or ".")
return root / "documentation-single" / "index.html"
def _article_root(soup: BeautifulSoup) -> Tag | BeautifulSoup:
return (
soup.select_one('[itemprop="articleBody"]')
or soup.select_one('[role="main"]')
or soup.select_one(".document")
or soup.body
or soup
)
def _load_article(path: Path) -> Tag | BeautifulSoup:
html = path.read_text(encoding="utf-8", errors="replace")
soup = BeautifulSoup(html, "html.parser")
for element in soup(_STRIP_TAGS):
element.decompose()
return _article_root(soup)
def _is_section_wrapper(tag: Tag) -> bool:
name = (tag.name or "").lower()
if name == "section":
return True
classes = tag.get("class") or []
return name == "div" and any(c in ("section", "toctree-wrapper") for c in classes)
[docs]
def parse_help_sections(path: Path) -> list[tuple[str, str]]:
"""Parse singlehtml into ``(heading_path, body_text)`` sections."""
return _iter_sections(_load_article(path))
def _iter_sections(article: Tag | BeautifulSoup) -> list[tuple[str, str]]:
sections: list[tuple[str, str]] = []
stack: list[tuple[int, str]] = []
body_parts: list[str] = []
def flush() -> None:
nonlocal body_parts
title = " › ".join(t for _, t in stack) if stack else "(preamble)"
text = "\n".join(body_parts).strip()
if text or (stack and title != "(preamble)"):
sections.append((title, text))
body_parts = []
def push_heading(level: int, title: str) -> None:
flush()
while stack and stack[-1][0] >= level:
stack.pop()
stack.append((level, title))
def walk(node: Tag | BeautifulSoup) -> None:
for child in getattr(node, "children", []):
if isinstance(child, NavigableString):
continue
if not isinstance(child, Tag):
continue
name = (child.name or "").lower()
if name in _HEADING_TAGS:
level = int(name[1])
title = child.get_text(" ", strip=True) or name
push_heading(level, title)
continue
if _is_section_wrapper(child):
walk(child)
continue
if name in _BLOCK_TAGS:
text = child.get_text(" ", strip=True)
if text:
body_parts.append(text)
continue
if child.find(_HEADING_TAGS) or child.find(_is_section_wrapper):
walk(child)
else:
text = child.get_text(" ", strip=True)
if text and name not in ("a", "span", "em", "strong", "code"):
body_parts.append(text)
walk(article)
flush()
return sections
def _toc(sections: list[tuple[str, str]]) -> str:
headings = [title for title, _ in sections if title != "(preamble)"]
if not headings:
return "(no headings found)"
seen: set[str] = set()
lines: list[str] = []
for h in headings:
if h in seen:
continue
seen.add(h)
lines.append(f"- {h}")
return "\n".join(lines)
def _truncate(text: str, max_chars: int) -> str:
if max_chars <= 0 or len(text) <= max_chars:
return text
return text[:max_chars] + "\n... [Content truncated]"
def _default_max_chars() -> int:
ctx = int(getattr(settings, "LLM_CONTEXT_WINDOW", 32768))
return max(8000, min(ctx * 2, 96000))
[docs]
def read_agency_help_text(
query: str = "",
max_chars: int | None = None,
*,
path: Path | None = None,
k: int = 6,
) -> str:
"""Core implementation (testable without LangChain invoke)."""
if max_chars is None:
max_chars = _default_max_chars()
doc_path = path or singlehtml_path()
if not doc_path.is_file():
return (
f"Error: Agency help singlehtml not found at {doc_path}. "
"Build with `./run document-single` in agency-server."
)
try:
sections = parse_help_sections(doc_path)
except OSError as exc:
return f"Error reading help file: {exc}"
except Exception as exc: # pragma: no cover - parser edge cases
return f"Error parsing help HTML: {exc}"
if not query.strip():
return _truncate(
"Agency documentation (Sphinx singlehtml) — table of contents.\n"
"Call again with a natural-language `query` to retrieve matching sections.\n\n"
f"{_toc(sections)}",
max_chars,
)
try:
from agency.tools.help_index import search_agency_docs
hits = search_agency_docs(query.strip(), k=k, path=doc_path)
except Exception as exc:
return (
f"Error: semantic help search failed ({exc}). "
"Ensure ChromaDB and Ollama embeddings are reachable, "
"then run `manage.py index_agency_docs` or `./run document-single`."
)
if not hits:
return (
f"No sections matched query {query!r}. "
"Try rephrasing, or call without query to list headings."
)
chunks = [
(f"## {title}\n{body}".strip() if body else f"## {title}")
for title, body, _score in hits
]
return _truncate("\n\n".join(chunks), max_chars)
@tool()
def read_agency_help(query: str = "", max_chars: int = 0) -> str:
"""Read Agency product help from the built Sphinx singlehtml manual.
Same source as Mission Control Help (`/static/documentation-single/`).
Without a query, returns a compact table of contents. With a query,
runs semantic search over documentation sections (Spanish or English OK).
Args:
query: Natural-language question or topic (e.g. who created Agency,
Channels WebSocket, how missions work)
max_chars: Max characters to return (0 = derive from LLM_CONTEXT_WINDOW)
Returns:
TOC, matching documentation text, or an error if docs/index are unavailable
"""
return read_agency_help_text(
query=query,
max_chars=None if max_chars <= 0 else max_chars,
)