from __future__ import annotations import os import re import tempfile import time from collections import OrderedDict, deque from concurrent.futures import Future, ThreadPoolExecutor, as_completed from datetime import datetime from typing import Annotated, Callable, Dict, List, Tuple from urllib.parse import urlparse import gradio as gr import requests from bs4 import BeautifulSoup from ddgs import DDGS from huggingface_hub import InferenceClient from .Web_Fetch import _fullpage_markdown_from_soup, _http_get_enhanced from app import _log_call_end, _log_call_start, _search_rate_limiter, _truncate_for_log from ._docstrings import autodoc HF_TEXTGEN_TOKEN = os.getenv("HF_READ_TOKEN") or os.getenv("HF_TOKEN") # Single source of truth for the LLM-facing tool description TOOL_SUMMARY = ( "Write a summary of what the user wants to research, and " "run multiple DuckDuckGo searches (up to 50 max results between all queries), fetch pages, and a Research agent will produce a comprehensive research report with sources; " "returns (Markdown report, newline-separated source links, downloadable report path). " "Provide the user with one-paragraph summary of the research report and the txt file in this format `![research_report](URL)`." ) RESEARCHER_SYSTEM_PROMPT = ( "You are Nymbot, a helpful deep research assistant. You will be asked a Query from a user and you will create a long, comprehensive, well-structured research report in response to the user's Query.\n\n" "You will receive a summary of the user question, the search queries used, and the fetched webpages. Follow the guidance below when writing the report.\n\n" "\n" "Write a well-formatted report in the structure of a scientific report to a broad audience. The report must be readable and have a nice flow of Markdown headers and paragraphs of text. Do NOT use bullet points or lists which break up the natural flow. The report must be exhaustive for comprehensive topics.\n" "For any given user query, first determine the major themes or areas that need investigation, then structure these as main sections, and develop detailed subsections that explore various facets of each theme. Each section and subsection requires paragraphs of texts that need to all connect into one narrative flow.\n" "\n\n" "\n" "- Always begin with a clear title using a single # header\n" "- Organize content into major sections using ## headers\n" "- Further divide into subsections using ### headers\n" "- Use #### headers sparingly for special subsections\n" "- Never skip header levels\n" "- Write multiple paragraphs per section or subsection\n" "- Each paragraph must contain at least 4-5 sentences, present novel insights and analysis grounded in source material, connect ideas to original query, and build upon previous paragraphs to create a narrative flow\n" "- Never use lists, instead always use text or tables\n\n" "Mandatory Section Flow:\n" "1. Title (# level)\n - Before writing the main report, start with one detailed paragraph summarizing key findings\n" "2. Main Body Sections (## level)\n - Each major topic gets its own section (## level). There MUST BE at least 5 sections.\n - Use ### subsections for detailed analysis\n - Every section or subsection needs at least one paragraph of narrative before moving to the next section\n - Do NOT have a section titled \"Main Body Sections\" and instead pick informative section names that convey the theme of the section\n" "3. Conclusion (## level)\n - Synthesis of findings\n - Potential recommendations or next steps\n" "\n\n" "\n" "- Always break it down into multiple steps\n" "- Assess the different sources and whether they are useful for any steps needed to answer the query\n" "- Create the best report that weighs all the evidence from the sources\n" "- Use the current date supplied in the first user message to contextualize findings\n" "- Make sure that your final report addresses all parts of the query\n" "- Communicate a brief high-level plan in the introduction; do not reveal chain-of-thought.\n" "- When referencing sources during analysis, you should still refer to them by index with brackets and follow \n" "- As a final step, review your planned report structure and ensure it completely answers the query.\n" "\n\n" ) FILTERER_SYSTEM_PROMPT = ( "You are Nymbot Filterer, an analyst who selects the most relevant sources for a research task. " "You will be given a summary of the research topic (and optional search queries) followed by multiple fetched documents. " "Each document includes its URL and a truncated excerpt. Evaluate how well each source helps answer the research topic. " "Return only the URLs that should be used for the final research step. Output plain text with exactly one URL per line and no additional commentary, bullets, numbering, or explanations. " "If no sources are relevant, return an empty string." ) class SlowHost(Exception): pass def _normalize_query(q: str) -> str: if not q: return "" repl = {"“": '"', "”": '"', "‘": "'", "’": "'", "`": "'"} for key, value in repl.items(): q = q.replace(key, value) q = re.sub(r"\s+", " ", q) q = re.sub(r'"\s+"', " ", q) q = q.strip().strip('"').strip() return q def _search_urls_only(query: str, max_results: int) -> list[str]: if not query or not query.strip() or max_results <= 0: return [] urls: list[str] = [] try: _search_rate_limiter.acquire() with DDGS() as ddgs: for item in ddgs.text(query, region="wt-wt", safesearch="moderate", max_results=max_results): url = (item.get("href") or item.get("url") or "").strip() if url: urls.append(url) except Exception: pass seen = set() deduped = [] for url in urls: if url not in seen: seen.add(url) deduped.append(url) return deduped def _fetch_page_markdown_fast(url: str, max_chars: int = 3000, timeout: float = 10.0) -> str: try: resp = _http_get_enhanced(url, timeout=timeout, skip_rate_limit=True) resp.raise_for_status() except requests.exceptions.RequestException as exc: msg = str(exc) if "timed out" in msg.lower(): raise SlowHost(msg) from exc return "" final_url = str(resp.url) ctype = resp.headers.get("Content-Type", "") if "html" not in ctype.lower(): return "" resp.encoding = resp.encoding or resp.apparent_encoding html = resp.text soup = BeautifulSoup(html, "lxml") md_text = _fullpage_markdown_from_soup(soup, final_url, "") if max_chars > 0 and len(md_text) > max_chars: md_text = md_text[:max_chars] return md_text def _truncate_join(parts: List[str], max_chars: int) -> Tuple[str, bool]: out = [] total = 0 truncated = False for part in parts: if not part: continue if total + len(part) > max_chars: out.append(part[: max(0, max_chars - total)]) truncated = True break out.append(part) total += len(part) return ("\n\n".join(out), truncated) def _build_research_prompt(summary: str, queries: List[str], url_list: List[str], pages_map: Dict[str, str]) -> str: sources_blocks: List[str] = [] indexed_urls: List[str] = [] for idx, url in enumerate(url_list, start=1): text = pages_map.get(url, "").strip() if not text: continue indexed_urls.append(f"[{idx}] {url}") sources_blocks.append(f"[Source {idx}] URL: {url}\n\n{text}") sources_joined, truncated = _truncate_join(sources_blocks, max_chars=100_000) prompt_parts: List[str] = [] prompt_parts.append("\n" + (summary or "") + "\n\n") populated = [q for q in queries if q and q.strip()] if populated: prompt_parts.append("\n" + "\n".join(f"- {q.strip()}" for q in populated) + "\n\n") if indexed_urls: prompt_parts.append("\n" + "\n".join(indexed_urls) + "\n\n") prompt_parts.append("\n" + sources_joined + ("\n\n[NOTE] Sources truncated due to context limits." if truncated else "") + "\n") return "\n\n".join(prompt_parts) def _build_filter_prompt(summary: str, queries: List[str], pages_map: Dict[str, str]) -> str: populated = [q for q in queries if q and q.strip()] summary_text = summary or "" prompt_sections: List[str] = [] prompt_sections.append("\n" + summary_text + "\n") if populated: prompt_sections.append("\n" + "\n".join(populated) + "\n") sources: List[str] = [] for idx, (url, text) in enumerate(pages_map.items(), start=1): content = text.strip() if not content: continue sources.append(f"[Source {idx}] URL: {url}\n\n{content}") sources_joined, truncated = _truncate_join(sources, max_chars=60_000) prompt_sections.append("\n" + sources_joined + ("\n\n[NOTE] Sources truncated due to context limits." if truncated else "") + "\n") prompt_sections.append( "\nIdentify which of the provided URLs should be retained for the final research synthesis. " "Consider coverage, credibility, and relevance to the research topic. " "Return ONLY the URLs you choose, with one URL per line and no additional text.\n" ) return "\n\n".join(prompt_sections) def _parse_filterer_output(raw: str, allowed_urls: List[str]) -> List[str]: if not raw: return [] allowed_set = {url.strip(): idx for idx, url in enumerate(allowed_urls)} found_indices: set[int] = set() for line in raw.splitlines(): candidate = line.strip() if not candidate: continue if candidate in allowed_set: found_indices.add(allowed_set[candidate]) continue match = re.search(r"https?://[^\s]+", candidate) if not match: continue url = match.group(0).rstrip(".,);]") if url in allowed_set: found_indices.add(allowed_set[url]) selected = [allowed_urls[idx] for idx in sorted(found_indices)] return selected def _write_report_tmp(text: str) -> str: tmp_dir = tempfile.mkdtemp(prefix="deep_research_") path = os.path.join(tmp_dir, "research_report.txt") with open(path, "w", encoding="utf-8") as file: file.write(text) return path def _fetch_pages_within_budget(urls: List[str], char_limit: int, time_left_fn: Callable[[], float]) -> OrderedDict: pages: dict[str, str] = {} if not urls: return OrderedDict() queue = deque(urls) attempts: dict[str, int] = {url: 0 for url in urls} max_attempts = 2 max_workers = min(12, max(4, len(urls))) in_flight: dict[Future, str] = {} delayed: list[tuple[float, str]] = [] def schedule_next(executor: ThreadPoolExecutor) -> None: while queue and len(in_flight) < max_workers: url = queue.popleft() if url in pages: continue attempts.setdefault(url, 0) if attempts[url] >= max_attempts: continue attempts[url] += 1 tl = time_left_fn() if tl <= 0.1: return per_timeout = 10.0 if tl > 15 else (5.0 if tl > 8 else 2.0) future = executor.submit(_fetch_page_markdown_fast, url, char_limit, per_timeout) in_flight[future] = url with ThreadPoolExecutor(max_workers=max_workers) as executor: schedule_next(executor) while (in_flight or queue or delayed) and time_left_fn() > 0.2: now = time.time() if delayed: ready: list[tuple[float, str]] = [] not_ready: list[tuple[float, str]] = [] for ready_time, delayed_url in delayed: (ready if ready_time <= now else not_ready).append((ready_time, delayed_url)) delayed = not_ready for _, delayed_url in ready: queue.append(delayed_url) if ready: schedule_next(executor) done = [future for future in list(in_flight.keys()) if future.done()] if not done: if not queue and delayed: next_ready = min((t for t, _ in delayed), default=time.time()) sleep_for = max(0.0, next_ready - time.time()) time.sleep(max(0.02, min(0.25, sleep_for))) else: time.sleep(0.05) continue for future in done: url = in_flight.pop(future) try: md = future.result() if md and not md.startswith("Unsupported content type") and not md.startswith("An error occurred"): pages[url] = md try: print(f"[FETCH OK] {url} (chars={len(md)})", flush=True) except Exception: pass except SlowHost: if time_left_fn() > 5.0: delayed.append((time.time() + 3.0, url)) except Exception: pass schedule_next(executor) ordered = OrderedDict((url, pages[url]) for url in urls if url in pages) return ordered @autodoc( summary=TOOL_SUMMARY, ) def Deep_Research( summary: Annotated[str, "Summarization of research topic (one or more sentences)."], query1: Annotated[str, "DDG Search Query 1"], max1: Annotated[int, "Max results for Query 1 (1-50)"] = 10, query2: Annotated[str, "DDG Search Query 2"] = "", max2: Annotated[int, "Max results for Query 2 (1-50)"] = 10, query3: Annotated[str, "DDG Search Query 3"] = "", max3: Annotated[int, "Max results for Query 3 (1-50)"] = 10, query4: Annotated[str, "DDG Search Query 4"] = "", max4: Annotated[int, "Max results for Query 4 (1-50)"] = 10, query5: Annotated[str, "DDG Search Query 5"] = "", max5: Annotated[int, "Max results for Query 5 (1-50)"] = 10, ) -> tuple[str, str, str]: _log_call_start( "Deep_Research", summary=_truncate_for_log(summary or "", 200), queries=[q for q in [query1, query2, query3, query4, query5] if q], ) if not HF_TEXTGEN_TOKEN: _log_call_end("Deep_Research", "error=missing HF token") raise gr.Error("Please provide a `HF_READ_TOKEN` to enable Deep Research.") queries = [ _normalize_query(query1 or ""), _normalize_query(query2 or ""), _normalize_query(query3 or ""), _normalize_query(query4 or ""), _normalize_query(query5 or ""), ] reqs = [ max(1, min(50, int(max1))), max(1, min(50, int(max2))), max(1, min(50, int(max3))), max(1, min(50, int(max4))), max(1, min(50, int(max5))), ] total_requested = sum(reqs) if total_requested > 50: reqs = [10, 10, 10, 10, 10] start_ts = time.time() budget_seconds = 55.0 deadline = start_ts + budget_seconds def time_left() -> float: return max(0.0, deadline - time.time()) now_dt = datetime.now().astimezone() date_str = now_dt.strftime("%A, %B %d, %Y %I:%M %p %Z").strip() if not date_str: date_str = now_dt.isoformat() all_urls: list[str] = [] tasks = [] with ThreadPoolExecutor(max_workers=min(5, sum(1 for q in queries if q.strip())) or 1) as executor: for query, count in zip(queries, reqs): if not query.strip(): continue tasks.append(executor.submit(_search_urls_only, query.strip(), count)) for future in as_completed(tasks): try: urls = future.result() or [] except Exception: urls = [] for url in urls: if url not in all_urls: all_urls.append(url) if len(all_urls) >= 50: break if time_left() <= 0.5: break if len(all_urls) > 50: all_urls = all_urls[:50] blacklist = { "homedepot.com", "tractorsupply.com", "mcmaster.com", "mrchain.com", "answers.com", "city-data.com", "dictionary.cambridge.org", } def _domain(url: str) -> str: try: return urlparse(url).netloc.lower() except Exception: return "" all_urls = [url for url in all_urls if _domain(url) not in blacklist] skip_exts = ( ".pdf", ".ppt", ".pptx", ".doc", ".docx", ".xls", ".xlsx", ".zip", ".gz", ".tgz", ".bz2", ".7z", ".rar", ) def _skip_url(url: str) -> bool: try: path = urlparse(url).path.lower() except Exception: return False return any(path.endswith(ext) for ext in skip_exts) all_urls = [url for url in all_urls if not _skip_url(url)] truncated_pages = OrderedDict() if all_urls and time_left() > 0.2: truncated_pages = _fetch_pages_within_budget(all_urls, 3000, time_left) print( f"[PIPELINE] Initial fetch complete: candidates={len(all_urls)}, truncated_documents={len(truncated_pages)}, time_left={time_left():.2f}s", flush=True, ) def _invoke_chat(messages, provider: str, max_tokens: int, temp: float, top_p: float): client = InferenceClient(provider=provider, api_key=HF_TEXTGEN_TOKEN) return client.chat.completions.create( model="Qwen/Qwen3-235B-A22B-Thinking-2507", messages=messages, max_tokens=max_tokens, temperature=temp, top_p=top_p, ) filtered_urls: List[str] = list(truncated_pages.keys()) filter_output = "" filter_used_fallback = False filter_success = False if truncated_pages and time_left() > 3.0: filter_prompt = _build_filter_prompt(summary or "", [q for q in queries if q.strip()], truncated_pages) filter_messages = [ {"role": "system", "content": FILTERER_SYSTEM_PROMPT}, {"role": "user", "content": f"The current date is {date_str}. Consider how recent each source is when deciding relevance."}, {"role": "user", "content": filter_prompt}, ] filter_completion = None try: print("[FILTER] Attempt 1: provider=cerebras, max_tokens=2048", flush=True) filter_completion = _invoke_chat(filter_messages, "cerebras", 2048, 0.2, 0.9) except Exception as exc1: print(f"[FILTER] Attempt 1 failed: {str(exc1)[:200]}", flush=True) try: print("[FILTER] Attempt 2: provider=auto, max_tokens=2048", flush=True) filter_completion = _invoke_chat(filter_messages, "auto", 2048, 0.2, 0.9) except Exception as exc2: print(f"[FILTER] Attempt 2 failed: {str(exc2)[:200]}", flush=True) if filter_completion and filter_completion.choices: filter_output = filter_completion.choices[0].message.content or "" filtered_urls = _parse_filterer_output(filter_output, list(truncated_pages.keys())) filter_success = bool(filter_output.strip()) and bool(filtered_urls) if not filtered_urls: filter_used_fallback = True fallback_count = min(8, len(truncated_pages)) filtered_urls = list(truncated_pages.keys())[:fallback_count] max_final_urls = 20 if len(filtered_urls) > max_final_urls: filter_used_fallback = True filtered_urls = filtered_urls[:max_final_urls] if not filter_success: filter_used_fallback = True print( f"[FILTER] Selected URLs={len(filtered_urls)}, fallback={filter_used_fallback}, time_left={time_left():.2f}s", flush=True, ) final_pages_fetched = OrderedDict() if filtered_urls and time_left() > 0.2: final_pages_fetched = _fetch_pages_within_budget(filtered_urls, 8000, time_left) merged_pages = OrderedDict() for url in filtered_urls: content = final_pages_fetched.get(url) or truncated_pages.get(url) or "" if content: merged_pages[url] = content pages = merged_pages print( f"[PIPELINE] Final fetch complete: retained_documents={len(pages)}, time_left={time_left():.2f}s", flush=True, ) prompt = _build_research_prompt(summary=summary or "", queries=[q for q in queries if q.strip()], url_list=list(pages.keys()), pages_map=pages) system_message = {"role": "system", "content": RESEARCHER_SYSTEM_PROMPT} date_message = {"role": "user", "content": f"The current date is {date_str}. Return only the research report."} messages = [ system_message, date_message, {"role": "user", "content": prompt}, ] try: prompt_chars = len(prompt) except Exception: prompt_chars = -1 print(f"[PIPELINE] Fetch complete: pages={len(pages)}, unique_urls={len(pages.keys())}, prompt_chars={prompt_chars}", flush=True) print("[PIPELINE] Starting inference (provider=cerebras, model=Qwen/Qwen3-235B-A22B-Thinking-2507)", flush=True) try: print("[LLM] Attempt 1: provider=cerebras, max_tokens=32768", flush=True) completion = _invoke_chat(messages, "cerebras", max_tokens=32768, temp=0.3, top_p=0.95) except Exception as exc1: print(f"[LLM] Attempt 1 failed: {str(exc1)[:200]}", flush=True) try: prompt2 = _build_research_prompt( summary=summary or "", queries=[q for q in queries if q.strip()], url_list=list(pages.keys())[:30], pages_map={key: pages[key] for key in list(pages.keys())[:30]}, ) messages = [ system_message, date_message, {"role": "user", "content": prompt2}, ] print("[LLM] Attempt 2: provider=cerebras (trimmed), max_tokens=16384", flush=True) completion = _invoke_chat(messages, "cerebras", max_tokens=16384, temp=0.7, top_p=0.95) except Exception as exc2: print(f"[LLM] Attempt 2 failed: {str(exc2)[:200]}", flush=True) try: print("[LLM] Attempt 3: provider=auto, max_tokens=8192", flush=True) completion = _invoke_chat(messages, "auto", max_tokens=8192, temp=0.7, top_p=0.95) except Exception as exc3: _log_call_end("Deep_Research", f"error={_truncate_for_log(str(exc3), 260)}") raise gr.Error(f"Researcher model call failed: {exc3}") raw = completion.choices[0].message.content or "" try: no_think = re.sub(r"[\s\S]*?<\\/think>", "", raw, flags=re.IGNORECASE) no_think = re.sub(r"<\\/?think>", "", no_think, flags=re.IGNORECASE) except Exception: no_think = raw try: paragraphs = [p for p in re.split(r"\n\s*\n", no_think) if p.strip()] keep: List[str] = [] removed = 0 planning_re = re.compile(r"\b(let me|now i(?:'ll| will)?|first,|i will now|i will|i'll|let's|now let me|i need to|now i'll|now i will)\b", re.IGNORECASE) for paragraph in paragraphs: if planning_re.search(paragraph): removed += 1 continue keep.append(paragraph) report = "\n\n".join(keep).strip() if not report: report = no_think.strip() except Exception: report = no_think removed = 0 report = re.sub(r"\n\s*\n\s*\n+", "\n\n", report) try: print(f"[POSTPROCESS] removed_planning_paragraphs={removed}, raw_chars={len(raw)}, final_chars={len(report)}", flush=True) except Exception: pass links_text = "\n".join([f"[{i+1}] {url}" for i, url in enumerate(pages.keys())]) if links_text: sources_section = "\n\n## Sources\n" + "\n".join([f"[{i+1}] {url}" for i, url in enumerate(pages.keys())]) report = report.rstrip() + sources_section file_path = _write_report_tmp(report) elapsed = time.time() - start_ts print(f"[TIMING] Deep_Research elapsed: {elapsed:.2f}s", flush=True) _log_call_end("Deep_Research", f"urls={len(pages)} file={os.path.basename(file_path)} duration={elapsed:.2f}s") return report, links_text, file_path def build_interface() -> gr.Interface: return gr.Interface( fn=Deep_Research, inputs=[ gr.Textbox(label="Summarization of research topic", lines=3, placeholder="Briefly summarize the research topic or user question"), gr.Textbox(label="DDG Search Query 1", max_lines=1), gr.Slider(1, 50, value=10, step=1, label="Max results (Q1)"), gr.Textbox(label="DDG Search Query 2", value="", max_lines=1), gr.Slider(1, 50, value=10, step=1, label="Max results (Q2)"), gr.Textbox(label="DDG Search Query 3", value="", max_lines=1), gr.Slider(1, 50, value=10, step=1, label="Max results (Q3)"), gr.Textbox(label="DDG Search Query 4", value="", max_lines=1), gr.Slider(1, 50, value=10, step=1, label="Max results (Q4)"), gr.Textbox(label="DDG Search Query 5", value="", max_lines=1), gr.Slider(1, 50, value=10, step=1, label="Max results (Q5)"), ], outputs=[ gr.Markdown(label="Research Report"), gr.Textbox(label="Fetched Links", lines=8), gr.File(label="Download Research Report", file_count="single"), ], title="Deep Research", description=( "
Perform multi-query web research: search with DuckDuckGo, fetch up to 50 pages in parallel, " "and generate a comprehensive report using a large LLM via Hugging Face Inference Providers (Cerebras). Requires HF_READ_TOKEN.
" ), api_description=TOOL_SUMMARY, flagging_mode="never", show_api=bool(HF_TEXTGEN_TOKEN), ) __all__ = ["Deep_Research", "build_interface"]