""" Site focus: read ONE site deeply instead of searching the whole web. Two ways this mode engages: 1. **Explicit** - the query names a host: "lyrics for X on genius.com" -> focus genius.com "site:stackoverflow.com asyncio gather" 2. **Implicit follow-up** - the previous turn already found the right page and the user now asks for "the full thing". A fresh web search would scatter again; instead we return to the best source from last turn and read it exhaustively. turn 1: "lyrics of Bohemian Rhapsody" -> answer + sources turn 2: "give me the full lyrics" -> re-read source [1] in full The second case is what makes multi-turn feel intelligent. A first search is necessarily shallow - snippets and partial extracts. The follow-up converts that into a complete answer from the page that actually holds the content. """ from __future__ import annotations import re from typing import Any, Dict, List, Optional from urllib.parse import urlparse # -------------------------------------------------------------------------- # 1. explicit site targeting # -------------------------------------------------------------------------- _SITE_OP = re.compile(r"\bsite:\s*([a-z0-9][-a-z0-9.]*\.[a-z]{2,})", re.I) _URL_IN_Q = re.compile(r"https?://([a-z0-9][-a-z0-9.]*\.[a-z]{2,})[^\s]*", re.I) _ON_SITE = re.compile( r"\b(?:on|from|at|using|via|check|read|open|scrape|crawl)\s+" r"(?:the\s+)?(?:site\s+)?([a-z0-9][-a-z0-9]*\.(?:[a-z]{2,}\.)?[a-z]{2,})\b", re.I, ) _NOT_A_DOMAIN = {"e.g", "i.e", "u.s", "u.k", "a.m", "p.m", "no.1"} _COMMON_TLD = ( ".com", ".org", ".net", ".io", ".ai", ".co", ".dev", ".me", ".tv", ".fm", ".info", ".gov", ".edu", ".uk", ".in", ".de", ".fr", ".jp", ".ru", ".to", ".site", ".xyz", ".app", ".news", ".wiki", ".press", ".blog", ) def extract_site(query: str) -> Optional[str]: """Return a host when the query explicitly targets one, else None.""" for pat in (_SITE_OP, _URL_IN_Q, _ON_SITE): m = pat.search(query or "") if not m: continue host = m.group(1).lower().strip(".,;:!?") if host in _NOT_A_DOMAIN or len(host) < 5: continue if re.fullmatch(r"[\d.]+", host): # "3.5", "v1.2" continue if not host.endswith(_COMMON_TLD): continue return host[4:] if host.startswith("www.") else host return None def strip_site_tokens(query: str) -> str: """Remove the site directive so what remains is the real question.""" q = _SITE_OP.sub(" ", query or "") q = _URL_IN_Q.sub(" ", q) q = _ON_SITE.sub(" ", q) q = " ".join(q.split()).strip(" ,.-") return q or (query or "").strip() # -------------------------------------------------------------------------- # 2. follow-up detection # -------------------------------------------------------------------------- _FOLLOW_PAT = re.compile( r"\b(" r"full|complete|entire|whole|everything|the rest|rest of it|" r"continue|go on|more detail|in detail|expand|elaborate|deeper|longer|" r"read it|read that|read the page|read the site|open it|open that|" r"that site|that source|that page|that link|same site|same source|" r"first (?:link|source|result)|from there" r")\b", re.I, ) _IMPERATIVE = re.compile( r"^\s*(?:and\s+|now\s+|ok(?:ay)?\s+|so\s+)*" r"(?:give|show|get|fetch|read|print|paste|list|write)\b", re.I, ) def is_follow_up(query: str, history: List[Any]) -> bool: """True when this turn should continue from the previous turn's sources.""" if not history: return False if not any(getattr(m, "role", "") == "assistant" for m in history): return False q = (query or "").strip() if not q: return False words = q.split() if _FOLLOW_PAT.search(q) and len(words) <= 16: return True if _IMPERATIVE.match(q) and len(words) <= 8: return True # Bare continuations: "more", "and the rest", "full one" if len(words) <= 4 and not q.endswith("?"): return True return False # -------------------------------------------------------------------------- # 3. conversation memory # -------------------------------------------------------------------------- def last_user_question(history: List[Any]) -> str: for m in reversed(history): if getattr(m, "role", "") == "user": txt = (getattr(m, "content", "") or "").strip() if txt: return txt return "" def remembered_sources(history: List[Any]) -> List[Dict[str, Any]]: """Sources attached to the most recent assistant turn.""" for m in reversed(history): if getattr(m, "role", "") == "assistant" and getattr(m, "sources", None): return [s for s in m.sources if isinstance(s, dict) and s.get("url")] return [] def pick_focus_url(history: List[Any], hint: str = "") -> Optional[Dict[str, Any]]: """ Choose which remembered source to read fully. Priority: 1. a domain the user just named 2. a site that HOSTS the requested content type (lyrics, recipe, code…) 3. the highest-ranked source that was actually readable Step 2 matters: an article *about* a song ranks above a lyrics site in a normal web search, but it is the wrong target once the user asks for the words themselves. """ srcs = remembered_sources(history) if not srcs: return None hint_host = extract_site(hint) if hint else None if hint_host: for s in srcs: if hint_host in (s.get("domain") or "") or hint_host in s.get("url", ""): return s kind = detect_content_kind(hint, last_user_question(history)) hosts = preferred_hosts(kind) if hosts: for s in srcs: dom = (s.get("domain") or "").lower() if any(h in dom for h in hosts): return s readable = [s for s in srcs if s.get("read")] or srcs readable.sort(key=lambda s: (-(s.get("score") or 0), s.get("id", 99))) return readable[0] def merge_context(query: str, history: List[Any]) -> str: """ Build a standalone question from a terse follow-up. "give me the full lyrics" on its own is unsearchable; joined with the previous turn it becomes "lyrics of Bohemian Rhapsody - full lyrics". """ prev = last_user_question(history) q = (query or "").strip() if not prev: return q if len(q.split()) > 16: # already self-contained return q return f"{prev} — {q}" def domain_of_url(url: str) -> str: try: host = urlparse(url).netloc.lower() return host[4:] if host.startswith("www.") else host except Exception: return "" # -------------------------------------------------------------------------- # 4. content-aware target selection # -------------------------------------------------------------------------- # When the user wants verbatim content, the right source is the site that # HOSTS it, not the highest-ranked article discussing it. Turn 1 for # "Bohemian Rhapsody lyrics" surfaces magazines writing *about* the song; # the follow-up "give me the full lyrics" needs an actual lyrics site. CONTENT_KINDS = { "lyrics": { "triggers": ("lyric", "lyrics", "words to the song", "song text", "verse", "chorus"), "hosts": ("genius.com", "azlyrics.com", "lyrics.com", "musixmatch.com", "lyricsfreak.com", "songlyrics.com", "metrolyrics.com", "letras.com", "sonichits.com", "lyricstranslate.com"), "search": "{q} lyrics", }, "tutorial": { "triggers": ("tutorial", "guide", "walkthrough", "how to build", "how to set up", "how to install", "step by step", "getting started", "course", "lesson"), "hosts": (), "search": "{q} tutorial guide", }, "article": { "triggers": ("full article", "whole article", "full post", "entire page", "full text", "read the whole"), "hosts": (), "search": "{q}", }, "recipe": { "triggers": ("recipe", "ingredients", "how to cook", "how to bake"), "hosts": ("allrecipes.com", "seriouseats.com", "bbcgoodfood.com", "food.com", "epicurious.com", "delish.com", "tasty.co"), "search": "{q} recipe ingredients method", }, "code": { "triggers": ("source code", "full code", "implementation", "snippet", "example code"), "hosts": ("github.com", "gitlab.com", "stackoverflow.com", "gist.github.com", "raw.githubusercontent.com"), "search": "{q} code example", }, "docs": { "triggers": ("documentation", "api reference", "official docs", "spec", "specification"), "hosts": ("docs.python.org", "developer.mozilla.org", "readthedocs.io", "docs.rs", "pkg.go.dev"), "search": "{q} official documentation", }, "transcript": { "triggers": ("transcript", "subtitles", "captions", "full speech", "what was said"), "hosts": ("rev.com", "scrapsfromtheloft.com", "subslikescript.com", "springfieldspringfield.co.uk", "genius.com"), "search": "{q} full transcript", }, "poem": { "triggers": ("poem", "poems", "stanza"), "hosts": ("poetryfoundation.org", "poets.org", "allpoetry.com", "gutenberg.org"), "search": "{q} full poem text", }, } def detect_content_kind(*texts): """Identify a verbatim-content request from the query and history.""" blob = " ".join(t.lower() for t in texts if t) for kind, spec in CONTENT_KINDS.items(): if any(tr in blob for tr in spec["triggers"]): return kind return None def preferred_hosts(kind): return CONTENT_KINDS.get(kind or "", {}).get("hosts", ()) def content_search_query(kind, q): tmpl = CONTENT_KINDS.get(kind or "", {}).get("search") return tmpl.format(q=q) if tmpl else q