""" TeCoxBeta orchestrator. plan -> fan-out search -> fuse -> rank -> (read || rerank) -> [multi-hop gap fill] -> chunk -> parallel subagents -> synthesize Every stage is deadline-aware and degrades gracefully: a dead provider, a slow page or an unavailable model reduces quality but never fails the request. Page reading and LLM reranking run CONCURRENTLY (one is network-bound, the other model-bound) which is a large latency win. """ from __future__ import annotations import asyncio import logging from typing import Any, AsyncGenerator, Dict, List, Optional, Sequence, Tuple from tecox.brain import ( brain, extract_key_points, find_gaps, llm_rerank, plan_queries, suggest_follow_ups, synthesize, synthesize_focus, ) from tecox.rewind import current_ledger, start_request_ledger from .config import DEPTH_PROFILES, DepthProfile, settings from .core import ( TTLCache, answer_cache, breaker, deadline_in, gather_capped, page_cache, serp_cache, ) from .focus import (content_search_query, detect_content_kind, extract_site, is_follow_up, last_user_question, merge_context, pick_focus_url, preferred_hosts, strip_site_tokens) from .providers import active_providers, provider_report from .ranking import merge_results, rank_chunks, rank_documents from .reader import chunk_document, read_documents, read_site_deep from .schemas import ( Chunk, Document, RawResult, SearchRequest, SearchResponse, Timings, registrable, ) log = logging.getLogger("tecox.engine") def get_profile(depth: str) -> DepthProfile: return DEPTH_PROFILES.get((depth or "fast").lower(), DEPTH_PROFILES["fast"]) _ACADEMIC_HINTS = ( "study", "studies", "research", "paper", "papers", "trial", "clinical", "meta-analysis", "journal", "peer-reviewed", "arxiv", "doi", "citation", "literature", "evidence for", "scientific", "efficacy", "biology", "physics", "chemistry", "neuroscience", "genome", "protein", "disease", "treatment", "therapy", "drug", "symptom", "diagnosis", ) _CODE_HINTS = ( "python", "javascript", "typescript", "rust", "golang", " java ", "c++", "error", "exception", "traceback", "bug", "install", "npm", "pip", "docker", "kubernetes", "api", "sdk", "library", "framework", "github", "code", "function", "compile", "debug", "regex", "sql", "git ", ) _SOCIAL_HINTS = ( "reddit", "opinion", "opinions", "people think", "discussion", "hacker news", "community", "experience", "worth it", "vs ", "recommend", "recommendation", "review", "reviews", "best ", ) _NEWS_HINTS = ( "news", "latest", "today", "yesterday", "this week", "current", "breaking", "announced", "launch", "price", "stock", "rate", "election", "score", "won", "winner", "update", "2026", "2025", ) _ADULT_HINTS = ( "porn", "porno", "xxx", "nsfw", "nude", "nudes", "naked", "sex ", " sex", "sexual", "erotic", "erotica", "hentai", "rule34", "camgirl", "cam girl", "onlyfans", "escort", "milf", "anal", "blowjob", "creampie", "orgasm", "masturbat", "fetish", "bdsm", "kink", "adult video", "adult site", "adult movie", "adult film", "xvideos", "pornhub", "xhamster", "redtube", "eporner", "brazzers", "hookup", "sexting", "strip club", "lingerie", "boobs", "tits", "pussy", "dick ", "cock ", "threesome", "gangbang", "amateur video", "jav ", "doujin", "ecchi", "cumshot", "handjob", ) _FILE_HINTS = ( "download", "torrent", "magnet", "repack", "iso ", " iso", "crack", "cracked", "free download", "full version", "rip", "portable", "setup", "installer", "seeders", "1080p", "720p", "x264", "x265", "bluray", "dvdrip", "webrip", "flac", "discography", "pdf download", "epub", "fitgirl", "dodi", "codex", "skidrow", "plaza", "empress", ) def is_file_query(query: str) -> bool: """Detect download intent so the file tier leads the fan-out.""" q = f" {query.lower()} " return any(w in q for w in _FILE_HINTS) def is_adult_query(query: str) -> bool: """Detect explicit intent so NSFW sources are engaged automatically.""" q = f" {query.lower()} " return any(w in q for w in _ADULT_HINTS) def _route_kinds(query: str, safe: bool = True) -> Optional[List[str]]: """ Intent routing: bias the provider pool toward sources that matter for this question. Web + reference are always in play; specialist pools are added only on evidence, because e.g. firing Crossref at "current RBI repo rate" floods the candidate set with irrelevant DOI landing pages. """ q = f" {query.lower()} " # Download intent is checked FIRST. "download gta san andreas" is a file # query, not an adult one - routing it to the adult tier returned porn # results for a game download, which is simply wrong. if is_file_query(query) and not is_adult_query(query): return ["files", "web", "archive", "social"] # Explicit queries: mainstream indexes return scrubbed junk for these, # so hand the query to sources that actually serve it. Nothing else is # dropped - general web stays in the pool for context and reviews. if is_adult_query(query): base = ["adult", "web", "social"] return base if safe else base + ["deep", "files", "archive"] kinds = {"web", "reference", "news"} if not safe: # Unrestricted: open the deep/onion surface, archives, files and # social. The adult tier is added only on actual adult intent - # blanket-adding it polluted ordinary queries with porn results. kinds |= {"deep", "archive", "files", "social"} if any(w in q for w in _ACADEMIC_HINTS): kinds.add("academic") if any(w in q for w in _CODE_HINTS): kinds.add("code") if any(w in q for w in _SOCIAL_HINTS): kinds.add("social") if any(w in q for w in _NEWS_HINTS) and "academic" not in kinds: kinds.discard("academic") return sorted(kinds) class DeepSearchEngine: """Reusable engine instance. Safe for concurrent requests.""" # ================================================================== # non-streaming # ================================================================== async def search(self, req: SearchRequest) -> SearchResponse: t = Timings() start_request_ledger() profile = get_profile(req.depth) query = req.query.strip() # `safe=false` (or explicit intent) unlocks the unrestricted path. # A filtered engine would refuse or sanitise the answer even when # retrieval succeeded, so the two must move together. req.unfiltered = (not req.safe) or is_adult_query(query) # ---- focus mode: one site, read deeply -------------------------- focus = self.resolve_focus(req) if focus: return await self._focus_search(req, focus, t) # ---- cache ------------------------------------------------------ # `model` (the capability tier) MUST be part of the key - without it # a TeF answer would be served back for a TeM request. ckey = TTLCache.key("answer", query.lower(), req.depth, req.model, req.max_results, req.answer, req.language, req.unfiltered, tuple(req.include_domains), tuple(req.exclude_domains), req.freshness) if req.cache: hit = answer_cache.get(ckey) if hit is not None: resp = SearchResponse(**hit) resp.cached = True resp.elapsed_ms = round(t.now(), 1) return resp # ---- 1+2. plan and FIRST search wave run CONCURRENTLY ------------ # The literal query is always worth searching, so fire it the moment # the request lands instead of idling through planner latency. if profile.planner and profile.sub_queries > 1 and brain.available: plan_task = asyncio.create_task( plan_queries(query, profile.sub_queries, model=req.model)) seed_task = asyncio.create_task( self._fanout(query, [query], profile, req)) queries = await plan_task t.mark("plan") seed_docs, seed_stats = await seed_task extra = [q for q in queries if q.strip().lower() != query.strip().lower()] if extra: docs, prov_stats = await self._fanout(query, extra, profile, req) docs = self._merge_docs(seed_docs, docs) for k, v in seed_stats.get("results_per_provider", {}).items(): prov_stats["results_per_provider"][k] = \ prov_stats["results_per_provider"].get(k, 0) + v prov_stats["raw_results"] += seed_stats.get("raw_results", 0) prov_stats["live_providers"] = sum( 1 for v in prov_stats["results_per_provider"].values() if v > 0) prov_stats["unique_urls"] = len(docs) else: docs, prov_stats = seed_docs, seed_stats else: queries = [query] t.mark("plan") docs, prov_stats = await self._fanout(query, queries, profile, req) t.mark("search") if not docs: return SearchResponse( query=query, depth=profile.name, answer=("No results could be retrieved. Every search backend is " "currently unreachable or rate-limited from this host."), sub_queries=queries, provider_stats=prov_stats, timings_ms=t.marks, elapsed_ms=round(t.now(), 1), ) # ---- 3. first-pass ranking (cheap, pre-read) -------------------- read_budget = profile.pages_to_read if req.read_pages is None else ( profile.pages_to_read if req.read_pages else 0) shortlist_n = max(req.max_results * 3, read_budget * 2, 20) shortlist = rank_documents(docs, queries, limit=min(shortlist_n, len(docs)), per_domain=3) t.mark("rank1") # ---- 4+5. read pages AND rerank CONCURRENTLY -------------------- # Reading is network-bound, reranking is model-bound; running them # serially wasted seconds. Reranking scores titles/snippets, so it # does not need the page bodies. jobs: List[Any] = [] if read_budget > 0: jobs.append(read_documents( shortlist, limit=read_budget, deadline=deadline_in(max(3.0, profile.budget_seconds * 0.55)), )) if profile.llm_rerank and brain.available: jobs.append(llm_rerank(query, shortlist, top_k=min(40, len(shortlist)), model=req.model)) if jobs: await asyncio.gather(*jobs, return_exceptions=True) t.mark("read+rerank") final_docs = rank_documents(shortlist, queries, limit=req.max_results, per_domain=2) for i, d in enumerate(final_docs, 1): d.cite_id = i # ---- 6. evidence chunks ---------------------------------------- single = profile.name == "instant" chunks = self._collect_chunks(final_docs, queries, profile.max_chunks, single_batch=single) t.mark("chunk") # ---- 7. multi-hop gap filling ---------------------------------- hops = 0 if (req.follow_up and profile.name in ("deep", "extreme", "ultra") and brain.available and chunks): hops, extra_docs = await self._multi_hop(query, queries, chunks, profile, req) if extra_docs: merged = {d.url: d for d in shortlist} for d in extra_docs: merged.setdefault(d.url, d) shortlist = list(merged.values()) final_docs = rank_documents(shortlist, queries, limit=req.max_results, per_domain=2) for i, d in enumerate(final_docs, 1): d.cite_id = i chunks = self._collect_chunks(final_docs, queries, profile.max_chunks, single_batch=single) t.mark("multihop") # ---- 8. synthesize --------------------------------------------- answer, key_points, follow_ups = "", [], [] if req.answer and chunks and brain.available: answer = await synthesize(query, chunks, depth=profile.name, language=req.language, model=req.model, unfiltered=req.unfiltered) if answer and profile.name in ("deep", "extreme", "ultra"): kp_res, fu_res = await asyncio.gather( extract_key_points(query, answer, model=req.model), suggest_follow_ups(query, answer, model=req.model), return_exceptions=True, ) key_points = kp_res if isinstance(kp_res, list) else [] follow_ups = fu_res if isinstance(fu_res, list) else [] t.mark("synthesize") lg = current_ledger() resp = SearchResponse( query=query, depth=profile.name, answer=answer, key_points=key_points, follow_ups=follow_ups, sources=[d.to_source() for d in final_docs], # type: ignore[arg-type] sub_queries=queries, provider_stats=prov_stats, stats={ "candidates_found": len(docs), "shortlisted": len(shortlist), "pages_read": sum(1 for d in shortlist if d.fetched), "evidence_chunks": len(chunks), "unique_domains": len({registrable(d.domain) for d in final_docs}), "multi_hop_rounds": hops, "unfiltered": req.unfiltered, "words_read": sum(d.word_count for d in shortlist if d.fetched), }, tokens=(lg.snapshot() if lg else {}), timings_ms=t.marks, elapsed_ms=round(t.now(), 1), ) if req.cache and (answer or final_docs): answer_cache.set(ckey, resp.model_dump(), settings.answer_cache_ttl) return resp # ================================================================== # focus mode: read one site deeply # ================================================================== @staticmethod def resolve_focus(req: SearchRequest) -> Optional[Dict[str, Any]]: """ Decide whether this request should read one site instead of searching. Returns {"site", "query", "seeds", "reason"} or None. """ # 1. caller named the site outright if req.site: return {"site": req.site.lower().lstrip("www."), "query": req.query, "seeds": [], "reason": "explicit"} # 2. the query names it: "lyrics on genius.com" host = extract_site(req.query) if host: return {"site": host, "query": strip_site_tokens(req.query), "seeds": [], "reason": "query"} # 3. terse follow-up -> reuse the source we cited last turn if req.messages and is_follow_up(req.query, req.messages): merged = merge_context(req.query, req.messages) kind = detect_content_kind(req.query, last_user_question(req.messages)) src = pick_focus_url(req.messages, req.query) hosts = preferred_hosts(kind) # If the remembered source does not HOST this kind of content, # go find one that does instead of re-reading an article about it. if kind and hosts: dom = ((src or {}).get("domain") or "").lower() if not any(h in dom for h in hosts): return {"site": "", "query": merged, "seeds": [], "reason": "follow_up", "kind": kind, "hosts": list(hosts)} if src and src.get("url", "").startswith("http"): return { "site": (src.get("domain") or src["url"].split("/")[2]).lstrip("www."), "query": merged, "seeds": [src["url"]], "reason": "follow_up", "kind": kind, } return None async def _focus_search(self, req: SearchRequest, focus: Dict[str, Any], t: Timings) -> SearchResponse: """Read one site exhaustively and answer from it alone.""" site, q = focus["site"], focus["query"] seeds = list(focus.get("seeds") or []) kind = focus.get("kind") hosts = list(focus.get("hosts") or []) # No site yet: find one that actually HOSTS this kind of content. if not site: probe = content_search_query(kind, q) sub = SearchRequest(query=probe, depth="instant", safe=req.safe, max_results=12, answer=False) docs, _ = await self._fanout(probe, [probe], get_profile("instant"), sub) best = None for d in docs: dom = (d.domain or "").lower() rank = next((i for i, h in enumerate(hosts) if h in dom), None) if rank is not None and (best is None or rank < best[0]): best = (rank, d) if best: site = best[1].domain.lstrip("www.") seeds = [best[1].url] elif docs: docs.sort(key=lambda d: -len(d.snippet)) site = docs[0].domain.lstrip("www.") seeds = [docs[0].url] else: return SearchResponse( query=req.query, depth=req.depth, answer="Could not locate a source hosting that content.", focus={"site": "", "mode": focus["reason"], "pages_read": 0}, timings_ms=t.marks, elapsed_ms=round(t.now(), 1)) # Site known but no entry point: search within it. if not seeds: sub = SearchRequest(query=f"{q} site:{site}", depth="instant", safe=req.safe, max_results=8, answer=False) # Several query shapes: engines widely ignore `site:`, so we also # try the bare host as a keyword. Without a real content-page seed # the crawler falls back to the homepage and finds nothing. probes = [f"{q} {site}", f"{site} {q}", f"site:{site} {q}", q] docs, _ = await self._fanout(q, probes, get_profile("instant"), sub) same = [d for d in docs if site in d.domain] same.sort(key=lambda d: (-len(d.url.rstrip("/").split("/")), -len(d.snippet))) seeds = [d.url for d in same[:4]] t.mark("locate") pages = await read_site_deep(site, q, max_pages=6, seed_urls=seeds) t.mark("read_site") # Some sites are JS-gated and unreadable even through the relay. # Falling back to a normal web search beats returning nothing - the # user asked a question, not for a status report about one host. if not pages: fallback = SearchRequest( query=q, depth=req.depth, safe=req.safe, model=req.model, max_results=req.max_results, answer=req.answer, language=req.language, unfiltered=req.unfiltered, cache=False) res = await engine.search(fallback) res.focus = {"site": site, "mode": focus["reason"], "pages_read": 0, "fallback": "site unreadable, searched the web instead"} res.elapsed_ms = round(t.now(), 1) return res for i, d in enumerate(pages, 1): d.cite_id = i d.score = 1.0 chunks: List[Chunk] = [] for d in pages: chunks.extend(chunk_document(d, target_chars=2400, max_chunks=20)) for c in chunks: c.score = 1.0 chunks = chunks[: max(24, get_profile(req.depth).max_chunks)] t.mark("chunk") answer = "" if req.answer and brain.available: answer = await synthesize_focus( q, chunks, site=site, depth=req.depth, language=req.language, model=req.model, unfiltered=req.unfiltered) t.mark("synthesize") lg = current_ledger() words = sum(d.word_count for d in pages) return SearchResponse( query=req.query, depth=req.depth, answer=answer, sources=[d.to_source() for d in pages], # type: ignore[arg-type] sub_queries=[q], focus={"site": site, "mode": focus["reason"], "pages_read": len(pages), "words_read": words}, stats={"candidates_found": len(pages), "pages_read": len(pages), "words_read": words, "unique_domains": 1, "evidence_chunks": len(chunks), "focused": True, "unfiltered": req.unfiltered}, tokens=(lg.snapshot() if lg else {}), timings_ms=t.marks, elapsed_ms=round(t.now(), 1)) # ------------------------------------------------------------------ async def _fanout( self, query: str, queries: Sequence[str], profile: DepthProfile, req: SearchRequest, on_provider: Optional[Any] = None, ) -> Tuple[List[Document], Dict[str, Any]]: kinds = _route_kinds(query, req.safe) if not req.providers else None providers = active_providers(names=req.providers or None, kinds=kinds) if not providers: providers = active_providers() kw: Dict[str, Any] = {} if req.freshness: kw["freshness"] = req.freshness if req.include_domains: kw["include_domains"] = req.include_domains if req.exclude_domains: kw["exclude_domains"] = req.exclude_domains if not req.safe: kw["explicit"] = True jobs: List[Tuple[str, str, Any]] = [] for q in queries: for p in providers: jobs.append((q, p.name, p.safe_search(q, profile.results_per_provider, on_event=on_provider, **kw))) deadline = deadline_in(profile.budget_seconds * 0.6) results = await gather_capped( (self._tagged(qq, pn, co) for qq, pn, co in jobs), limit=settings.search_concurrency, deadline=deadline, ) per_query: Dict[str, List[RawResult]] = {} counts: Dict[str, int] = {} for q, pname, rows in results: per_query.setdefault(q, []).extend(rows) counts[pname] = counts.get(pname, 0) + len(rows) docs = merge_results(per_query) if req.include_domains: inc = {d.lower().lstrip("www.") for d in req.include_domains} docs = [d for d in docs if any(d.domain.endswith(i) for i in inc)] if req.exclude_domains: exc = {d.lower().lstrip("www.") for d in req.exclude_domains} docs = [d for d in docs if not any(d.domain.endswith(e) for e in exc)] stats = { "queried": len(queries), "providers_used": len(providers), "results_per_provider": counts, "live_providers": sum(1 for v in counts.values() if v > 0), "raw_results": sum(counts.values()), "unique_urls": len(docs), "circuit_open": breaker.snapshot(), } return docs, stats @staticmethod def _merge_docs(a: List[Document], b: List[Document]) -> List[Document]: """Union two candidate sets, keeping the richer record per URL.""" by_url: Dict[str, Document] = {d.url: d for d in a} for d in b: prev = by_url.get(d.url) if prev is None: by_url[d.url] = d continue if len(d.title) > len(prev.title): prev.title = d.title if len(d.snippet) > len(prev.snippet): prev.snippet = d.snippet prev.providers.extend(d.providers) for pn, rk in d.ranks.items(): if pn not in prev.ranks or rk < prev.ranks[pn]: prev.ranks[pn] = rk for q in d.queries: if q not in prev.queries: prev.queries.append(q) return list(by_url.values()) @staticmethod async def _tagged(query: str, pname: str, coro: Any) -> Tuple[str, str, List[RawResult]]: return query, pname, (await coro or []) # ------------------------------------------------------------------ @staticmethod def _collect_chunks( docs: Sequence[Document], queries: Sequence[str], limit: int, single_batch: bool = False, ) -> List[Chunk]: """ Build the evidence pool. When `single_batch` is set (instant), trim so the whole context fits one prompt for the lowest possible latency. """ pool: List[Chunk] = [] for d in docs: pool.extend(chunk_document(d, max_chunks=6 if d.fetched else 1)) ranked = rank_chunks(pool, queries, limit) if not single_batch: return ranked budget = 12_000 kept: List[Chunk] = [] used = 0 for c in ranked: cost = len(c.text[:1500]) + 160 if used + cost > budget: continue kept.append(c) used += cost return kept or ranked[:3] # ------------------------------------------------------------------ async def _multi_hop( self, query: str, queries: List[str], chunks: List[Chunk], profile: DepthProfile, req: SearchRequest, ) -> Tuple[int, List[Document]]: max_rounds = 1 if profile.name == "deep" else 2 collected: List[Document] = [] rounds = 0 for _ in range(max_rounds): gaps = await find_gaps(query, chunks, max_gaps=3, model=req.model) gaps = [g for g in gaps if g.lower() not in {q.lower() for q in queries}][:3] if not gaps: break rounds += 1 queries.extend(gaps) sub = SearchRequest(query=query, depth=profile.name, providers=req.providers, freshness=req.freshness, include_domains=req.include_domains, exclude_domains=req.exclude_domains) docs, _ = await self._fanout(query, gaps, profile, sub) if not docs: break ranked = rank_documents(docs, gaps, limit=8, per_domain=2) await read_documents(ranked, limit=max(3, profile.pages_to_read // 3), deadline=deadline_in(profile.budget_seconds * 0.2)) collected.extend(ranked) for d in ranked: chunks.extend(chunk_document(d, max_chunks=3)) return rounds, collected # ================================================================== # streaming (Server-Sent Events) # ================================================================== async def stream_search( self, req: SearchRequest ) -> AsyncGenerator[Dict[str, Any], None]: """ Live pipeline stream. Every event is emitted at the moment the underlying work actually happens - each provider as it answers, each page as it is fetched and extracted, each agent as it reports. Nothing is a placeholder or a timed animation. An internal queue lets deep callbacks publish upwards while the pipeline keeps running. """ t = Timings() start_request_ledger() profile = get_profile(req.depth) query = req.query.strip() adult = is_adult_query(query) or (not req.safe) bus: asyncio.Queue = asyncio.Queue() async def emit(event: str, **data: Any) -> None: await bus.put({"event": event, "data": data}) async def drain() -> AsyncGenerator[Dict[str, Any], None]: while not bus.empty(): yield bus.get_nowait() async def run_stage(coro: Any) -> Any: """Await a stage while forwarding queued events in real time.""" task = asyncio.create_task(coro) while True: done, _ = await asyncio.wait({task}, timeout=0.05) while not bus.empty(): yield ("evt", bus.get_nowait()) if done: break yield ("result", task.result()) yield {"event": "start", "data": { "query": query, "depth": profile.name, "unfiltered": req.unfiltered, "explicit": adult}} # ---- plan ∥ first search wave ----------------------------------- yield {"event": "stage", "data": {"phase": "plan", "text": "Understanding your question"}} async def provider_cb(name: str, count: int, cached: bool) -> None: await emit("provider", name=name, results=count, cached=cached, ms=round(t.now(), 1)) seed_task = asyncio.create_task( self._fanout(query, [query], profile, req, on_provider=provider_cb)) # Planning and the seed wave run together. As soon as the planner # returns we launch the extra angles WITHOUT waiting for the seed # wave to finish, so no stage ever idles behind another. plan_task = ( asyncio.create_task(plan_queries(query, profile.sub_queries, model=req.model)) if (profile.planner and profile.sub_queries > 1 and brain.available) else None) if plan_task is not None: while not plan_task.done(): await asyncio.sleep(0.05) while not bus.empty(): yield bus.get_nowait() queries = plan_task.result() else: queries = [query] yield {"event": "plan", "data": {"sub_queries": queries, "ms": t.mark("plan")}} extra = [q for q in queries if q.strip().lower() != query.strip().lower()] fan_task = None if extra: yield {"event": "stage", "data": { "phase": "search", "text": f"Searching {len(extra)} more angles in parallel"}} fan_task = asyncio.create_task( self._fanout(query, extra, profile, req, on_provider=provider_cb)) pending = {tk for tk in (seed_task, fan_task) if tk is not None} while pending: done, pending = await asyncio.wait(pending, timeout=0.05) while not bus.empty(): yield bus.get_nowait() while not bus.empty(): yield bus.get_nowait() seed_docs, seed_stats = seed_task.result() if fan_task is not None: docs, prov_stats = fan_task.result() docs = self._merge_docs(seed_docs, docs) for k, v in seed_stats.get("results_per_provider", {}).items(): prov_stats["results_per_provider"][k] = \ prov_stats["results_per_provider"].get(k, 0) + v prov_stats["raw_results"] += seed_stats.get("raw_results", 0) prov_stats["live_providers"] = sum( 1 for v in prov_stats["results_per_provider"].values() if v > 0) prov_stats["unique_urls"] = len(docs) else: docs, prov_stats = seed_docs, seed_stats yield {"event": "search", "data": { "candidates": len(docs), "providers": prov_stats.get("live_providers", 0), "raw": prov_stats.get("raw_results", 0), "ms": t.mark("search")}} if not docs: yield {"event": "error", "data": {"message": "no results from any source"}} yield {"event": "done", "data": {"elapsed_ms": round(t.now(), 1)}} return # ---- rank -------------------------------------------------------- read_budget = profile.pages_to_read if req.read_pages is None else ( profile.pages_to_read if req.read_pages else 0) shortlist = rank_documents( docs, queries, limit=min(max(req.max_results * 3, 20), len(docs)), per_domain=3) yield {"event": "rank", "data": {"shortlist": len(shortlist), "ms": t.mark("rank1")}} # ---- read ∥ rerank, streaming each page as it lands -------------- async def page_cb(kind: str, d: Document) -> None: if kind == "start": await emit("reading", domain=d.domain, url=d.url, title=d.title[:120], ms=round(t.now(), 1)) else: await emit("read_done", domain=d.domain, url=d.url, words=d.word_count, ok=d.fetched, ms=round(t.now(), 1)) jobs: List[Any] = [] if read_budget > 0: yield {"event": "stage", "data": { "phase": "read", "text": f"Opening top {min(read_budget, len(shortlist))} sources"}} jobs.append(read_documents( shortlist, limit=read_budget, deadline=deadline_in(profile.budget_seconds * 0.55), on_event=page_cb)) if profile.llm_rerank and brain.available: jobs.append(llm_rerank(query, shortlist, top_k=min(40, len(shortlist)), model=req.model)) if jobs: # gather() already returns a future - wrapping it in # create_task() raises "a coroutine was expected". gtask = asyncio.gather(*jobs, return_exceptions=True) while not gtask.done(): await asyncio.sleep(0.05) while not bus.empty(): yield bus.get_nowait() while not bus.empty(): yield bus.get_nowait() yield {"event": "read", "data": { "pages_read": sum(1 for d in shortlist if d.fetched), "words": sum(d.word_count for d in shortlist if d.fetched), "reranked": bool(profile.llm_rerank and brain.available), "ms": t.mark("read+rerank")}} final_docs = rank_documents(shortlist, queries, limit=req.max_results, per_domain=2) for i, d in enumerate(final_docs, 1): d.cite_id = i yield {"event": "sources", "data": {"sources": [d.to_source() for d in final_docs]}} chunks = self._collect_chunks(final_docs, queries, profile.max_chunks, single_batch=profile.name == "instant") # ---- synthesize --------------------------------------------------- answer = "" if req.answer and chunks and brain.available: from tecox.brain import (DEPTH_STYLE, LEAD_CTX, SUBAGENT_CTX, SYNTH_SYS, UNFILTERED_SYS, depth_tokens, pack_context, run_subagents, split_waves, strip_reasoning) from tecox.rewind import rewind as _rw from tecox.rewind import catalog as _cat style = DEPTH_STYLE.get(profile.name, DEPTH_STYLE["fast"]) sys_prompt = UNFILTERED_SYS if req.unfiltered else SYNTH_SYS role, _w = _cat.tier_roles(req.model, req.unfiltered) if profile.name == "instant" and not req.unfiltered: role = "fast" total = sum(len(c.text[:2400]) for c in chunks) if total > LEAD_CTX: contexts = split_waves(chunks, SUBAGENT_CTX, max_agents=8) yield {"event": "stage", "data": { "phase": "agents", "text": f"{len(contexts)} agents cross-checking evidence"}} digests = await run_subagents(query, contexts, unfiltered=req.unfiltered, tier=req.model) findings = [d.strip() for d in digests if d and d.strip()] yield {"event": "agents", "data": {"count": len(findings), "ms": round(t.now(), 1)}} merged = "\n\n".join(f"### Agent {i + 1}\n{f}" for i, f in enumerate(findings))[:LEAD_CTX] prompt = ( f"VERIFIED FINDINGS from {len(findings)} parallel agents " f"(each fact carries its [n] citation):\n\n{merged}\n\n" f"QUESTION: {query}\n\n{style}\n" f"Merge duplicates. Preserve [n] citations exactly. Flag " f'contradictions.\nQUESTION (repeat): "{query}"') else: ctx = pack_context(chunks, LEAD_CTX) prompt = (f"SOURCES:\n{ctx}\n\nQUESTION: {query}\n\n{style}\n" f"Ground every claim with [n] citations.\n" f'QUESTION (repeat): "{query}"') yield {"event": "stage", "data": {"phase": "write", "text": "Writing the answer"}} buf: List[str] = [] async for tok in _rw.chat_stream( prompt, role=role, system=sys_prompt, max_tokens=depth_tokens(profile.name), timeout=360.0): buf.append(tok) yield {"event": "token", "data": {"t": tok}} answer = strip_reasoning("".join(buf).strip()) if not answer: answer = await synthesize(query, chunks, depth=profile.name, language=req.language, model=req.model, unfiltered=req.unfiltered) if answer: yield {"event": "token", "data": {"t": answer}} yield {"event": "answer", "data": {"answer": answer, "ms": t.mark("synthesize")}} if answer and profile.name in ("deep", "extreme", "ultra"): kp_res, fu_res = await asyncio.gather( extract_key_points(query, answer, model=req.model), suggest_follow_ups(query, answer, model=req.model), return_exceptions=True, ) yield {"event": "extras", "data": { "key_points": kp_res if isinstance(kp_res, list) else [], "follow_ups": fu_res if isinstance(fu_res, list) else []}} lg = current_ledger() yield {"event": "done", "data": { "elapsed_ms": round(t.now(), 1), "timings": t.marks, "tokens": (lg.snapshot() if lg else {}), "stats": { "candidates_found": len(docs), "pages_read": sum(1 for d in shortlist if d.fetched), "words_read": sum(d.word_count for d in shortlist if d.fetched), "unique_domains": len({registrable(d.domain) for d in final_docs}), "unfiltered": req.unfiltered, "explicit": adult, }, "provider_stats": prov_stats, }} engine = DeepSearchEngine() def health() -> Dict[str, Any]: return { "status": "ok", "providers": provider_report(), "cache": { "serp": serp_cache.stats(), "page": page_cache.stats(), "answer": answer_cache.stats(), }, "circuit_open": breaker.snapshot(), "depths": {k: {"sub_queries": v.sub_queries, "pages_to_read": v.pages_to_read, "budget_s": v.budget_seconds} for k, v in DEPTH_PROFILES.items()}, }