Refactoring the server to be a lot more module, easier to ingest

Modules/Deep_Research.py

@@ -0,0 +1,448 @@
+from __future__ import annotations
+
+import os
+import re
+import tempfile
+import time
+from collections import deque
+from concurrent.futures import Future, ThreadPoolExecutor, as_completed
+from typing import Annotated, 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
+
+HF_TEXTGEN_TOKEN = os.getenv("HF_READ_TOKEN") or os.getenv("HF_TOKEN")
+
+
+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:
+    researcher_instructions = (
+        "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 have been provided with User Question, Search Queries, and numerous webpages that the searches yielded.\n\n"
+        "<report_format>\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"
+        "</report_format>\n\n"
+        "<document_structure>\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"
+        "</document_structure>\n\n"
+        "<planning_rules>\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"
+        "- Remember that the current date is: Wednesday, April 23, 2025, 11:50 AM EDT\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 <citations>\n"
+        "- As a final step, review your planned report structure and ensure it completely answers the query.\n"
+        "</planning_rules>\n\n"
+    )
+    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 = [researcher_instructions]
+    prompt_parts.append("<user_query_summary>\n" + (summary or "") + "\n</user_query_summary>\n")
+    populated = [q for q in queries if q and q.strip()]
+    if populated:
+        prompt_parts.append("<search_queries>\n" + "\n".join(f"- {q.strip()}" for q in populated) + "\n</search_queries>\n")
+    if indexed_urls:
+        prompt_parts.append("<sources_list>\n" + "\n".join(indexed_urls) + "\n</sources_list>\n")
+    prompt_parts.append("<fetched_documents>\n" + sources_joined + ("\n\n[NOTE] Sources truncated due to context limits." if truncated else "") + "\n</fetched_documents>")
+    return "\n\n".join(prompt_parts)
+
+
+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 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())
+
+    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)]
+    pages: dict[str, str] = {}
+    if all_urls:
+        queue = deque(all_urls)
+        attempts: dict[str, int] = {url: 0 for url in all_urls}
+        max_attempts = 2
+        max_workers = min(12, max(4, len(all_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
+                if attempts[url] >= max_attempts:
+                    continue
+                attempts[url] += 1
+                tl = time_left()
+                per_timeout = 10.0 if tl > 15 else (5.0 if tl > 8 else 2.0)
+                future = executor.submit(_fetch_page_markdown_fast, url, 3000, per_timeout)
+                in_flight[future] = url
+
+        with ThreadPoolExecutor(max_workers=max_workers) as executor:
+            schedule_next(executor)
+            while (in_flight or queue) and time_left() > 0.2:
+                now = time.time()
+                if delayed:
+                    ready = []
+                    not_ready = []
+                    for ready_time, url in delayed:
+                        (ready if ready_time <= now else not_ready).append((ready_time, url))
+                    delayed = not_ready
+                    for _, url in ready:
+                        queue.append(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:
+                        sleep_for = max(0.02, min(0.25, max(0.0, min(t for t, _ in delayed) - time.time())))
+                        time.sleep(sleep_for)
+                    else:
+                        time.sleep(0.05)
+                else:
+                    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() > 5.0:
+                                delayed.append((time.time() + 3.0, url))
+                        except Exception:
+                            pass
+                    schedule_next(executor)
+    prompt = _build_research_prompt(summary=summary or "", queries=[q for q in queries if q.strip()], url_list=list(pages.keys()), pages_map=pages)
+    messages = [
+        {"role": "system", "content": "You are Nymbot, an expert deep research assistant."},
+        {"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)
+
+    def _run_inference(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,
+        )
+
+    try:
+        print("[LLM] Attempt 1: provider=cerebras, max_tokens=32768", flush=True)
+        completion = _run_inference("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 = [
+                {"role": "system", "content": "You are Nymbot, an expert deep research assistant."},
+                {"role": "user", "content": prompt2},
+            ]
+            print("[LLM] Attempt 2: provider=cerebras (trimmed), max_tokens=16384", flush=True)
+            completion = _run_inference("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 = _run_inference("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"<think>[\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())])
+    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=(
+            "<div style=\"text-align:center\">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.</div>"
+        ),
+        api_description=(
+            "Runs 1–5 DDG searches (URLs only), caps total results to 50 (when exceeding, each query returns 10). "
+            "Fetches all URLs (3000 chars each) and calls the Researcher to write a research report. "
+            "Returns the report (Markdown), the list of sources, and a downloadable text file path. "
+            "Provide the user with one-paragraph summary of the research report and the txt file in this format `![research_report.txt](URL)`"
+        ),
+        flagging_mode="never",
+        show_api=bool(HF_TEXTGEN_TOKEN),
+    )
+
+
+__all__ = ["Deep_Research", "build_interface"]

Modules/Generate_Image.py

@@ -0,0 +1,129 @@
+from __future__ import annotations
+
+import os
+import random
+from typing import Annotated
+
+import gradio as gr
+from PIL import Image
+from huggingface_hub import InferenceClient
+
+from app import _log_call_end, _log_call_start, _truncate_for_log
+
+HF_API_TOKEN = os.getenv("HF_READ_TOKEN")
+
+
+def Generate_Image(
+    prompt: Annotated[str, "Text description of the image to generate."],
+    model_id: Annotated[str, "Hugging Face model id in the form 'creator/model-name' (e.g., black-forest-labs/FLUX.1-Krea-dev)."] = "black-forest-labs/FLUX.1-Krea-dev",
+    negative_prompt: Annotated[str, "What should NOT appear in the image."] = (
+        "(deformed, distorted, disfigured), poorly drawn, bad anatomy, wrong anatomy, extra limb, "
+        "missing limb, floating limbs, (mutated hands and fingers), disconnected limbs, mutation, "
+        "mutated, ugly, disgusting, blurry, amputation, misspellings, typos"
+    ),
+    steps: Annotated[int, "Number of denoising steps (1–100). Higher = slower, potentially higher quality."] = 35,
+    cfg_scale: Annotated[float, "Classifier-free guidance scale (1–20). Higher = follow the prompt more closely."] = 7.0,
+    sampler: Annotated[str, "Sampling method label (UI only). Common options: 'DPM++ 2M Karras', 'DPM++ SDE Karras', 'Euler', 'Euler a', 'Heun', 'DDIM'."] = "DPM++ 2M Karras",
+    seed: Annotated[int, "Random seed for reproducibility. Use -1 for a random seed per call."] = -1,
+    width: Annotated[int, "Output width in pixels (64–1216, multiple of 32 recommended)."] = 1024,
+    height: Annotated[int, "Output height in pixels (64–1216, multiple of 32 recommended)."] = 1024,
+) -> Image.Image:
+    _log_call_start(
+        "Generate_Image",
+        prompt=_truncate_for_log(prompt, 200),
+        model_id=model_id,
+        steps=steps,
+        cfg_scale=cfg_scale,
+        seed=seed,
+        size=f"{width}x{height}",
+    )
+    if not prompt or not prompt.strip():
+        _log_call_end("Generate_Image", "error=empty prompt")
+        raise gr.Error("Please provide a non-empty prompt.")
+    enhanced_prompt = f"{prompt} | ultra detail, ultra elaboration, ultra quality, perfect."
+    providers = ["auto", "replicate", "fal-ai"]
+    last_error: Exception | None = None
+    for provider in providers:
+        try:
+            client = InferenceClient(api_key=HF_API_TOKEN, provider=provider)
+            image = client.text_to_image(
+                prompt=enhanced_prompt,
+                negative_prompt=negative_prompt,
+                model=model_id,
+                width=width,
+                height=height,
+                num_inference_steps=steps,
+                guidance_scale=cfg_scale,
+                seed=seed if seed != -1 else random.randint(1, 1_000_000_000),
+            )
+            _log_call_end("Generate_Image", f"provider={provider} size={image.size}")
+            return image
+        except Exception as exc:  # pylint: disable=broad-except
+            last_error = exc
+            continue
+    msg = str(last_error) if last_error else "Unknown error"
+    lowered = msg.lower()
+    if "404" in msg:
+        raise gr.Error(f"Model not found or unavailable: {model_id}. Check the id and your HF token access.")
+    if "503" in msg:
+        raise gr.Error("The model is warming up. Please try again shortly.")
+    if "401" in msg or "403" in msg:
+        raise gr.Error("Please duplicate the space and provide a `HF_READ_TOKEN` to enable Image and Video Generation.")
+    if ("api_key" in lowered) or ("hf auth login" in lowered) or ("unauthorized" in lowered) or ("forbidden" in lowered):
+        raise gr.Error("Please duplicate the space and provide a `HF_READ_TOKEN` to enable Image and Video Generation.")
+    _log_call_end("Generate_Image", f"error={_truncate_for_log(msg, 200)}")
+    raise gr.Error(f"Image generation failed: {msg}")
+
+
+def build_interface() -> gr.Interface:
+    return gr.Interface(
+        fn=Generate_Image,
+        inputs=[
+            gr.Textbox(label="Prompt", placeholder="Enter a prompt", lines=2),
+            gr.Textbox(
+                label="Model",
+                value="black-forest-labs/FLUX.1-Krea-dev",
+                placeholder="creator/model-name",
+                max_lines=1,
+                info="<a href=\"https://huggingface.co/models?pipeline_tag=text-to-image&inference_provider=nebius,cerebras,novita,fireworks-ai,together,fal-ai,groq,featherless-ai,nscale,hyperbolic,sambanova,cohere,replicate,scaleway,publicai,hf-inference&sort=trending\" target=\"_blank\" rel=\"noopener noreferrer\">Browse models</a>",
+            ),
+            gr.Textbox(
+                label="Negative Prompt",
+                value=(
+                    "(deformed, distorted, disfigured), poorly drawn, bad anatomy, wrong anatomy, extra limb, "
+                    "missing limb, floating limbs, (mutated hands and fingers), disconnected limbs, mutation, "
+                    "mutated, ugly, disgusting, blurry, amputation, misspellings, typos"
+                ),
+                lines=2,
+            ),
+            gr.Slider(minimum=1, maximum=100, value=35, step=1, label="Steps"),
+            gr.Slider(minimum=1.0, maximum=20.0, value=7.0, step=0.1, label="CFG Scale"),
+            gr.Radio(
+                label="Sampler",
+                value="DPM++ 2M Karras",
+                choices=["DPM++ 2M Karras", "DPM++ SDE Karras", "Euler", "Euler a", "Heun", "DDIM"],
+            ),
+            gr.Slider(minimum=-1, maximum=1_000_000_000, value=-1, step=1, label="Seed (-1 = random)"),
+            gr.Slider(minimum=64, maximum=1216, value=1024, step=32, label="Width"),
+            gr.Slider(minimum=64, maximum=1216, value=1024, step=32, label="Height"),
+        ],
+        outputs=gr.Image(label="Generated Image"),
+        title="Generate Image",
+        description=(
+            "<div style=\"text-align:center\">Generate images via Hugging Face serverless inference. "
+            "Default model is FLUX.1-Krea-dev.</div>"
+        ),
+        api_description=(
+            "Generate a single image from a text prompt using a Hugging Face model via serverless inference. "
+            "Supports creative prompts like 'a serene mountain landscape at sunset', 'portrait of a wise owl', "
+            "'futuristic city with flying cars'. Default model: FLUX.1-Krea-dev. "
+            "Parameters: prompt (str), model_id (str, creator/model-name), negative_prompt (str), steps (int, 1–100), "
+            "cfg_scale (float, 1–20), sampler (str), seed (int, -1=random), width/height (int, 64–1216). "
+            "Returns a PIL.Image. Return the generated media to the user in this format `![Alt text](URL)`"
+        ),
+        flagging_mode="never",
+        show_api=bool(os.getenv("HF_READ_TOKEN")),
+    )
+
+
+__all__ = ["Generate_Image", "build_interface"]

Modules/Generate_Speech.py

@@ -0,0 +1,164 @@
+from __future__ import annotations
+
+import numpy as np
+import gradio as gr
+
+from typing import Annotated
+
+from app import _log_call_end, _log_call_start, _truncate_for_log
+
+try:
+    import torch  # type: ignore
+except Exception:  # pragma: no cover
+    torch = None  # type: ignore
+
+try:
+    from kokoro import KModel, KPipeline  # type: ignore
+except Exception:  # pragma: no cover
+    KModel = None  # type: ignore
+    KPipeline = None  # type: ignore
+
+_KOKORO_STATE = {
+    "initialized": False,
+    "device": "cpu",
+    "model": None,
+    "pipelines": {},
+}
+
+
+def get_kokoro_voices() -> list[str]:
+    try:
+        from huggingface_hub import list_repo_files
+
+        files = list_repo_files("hexgrad/Kokoro-82M")
+        voice_files = [file for file in files if file.endswith(".pt") and file.startswith("voices/")]
+        voices = [file.replace("voices/", "").replace(".pt", "") for file in voice_files]
+        return sorted(voices) if voices else _get_fallback_voices()
+    except Exception:
+        return _get_fallback_voices()
+
+
+def _get_fallback_voices() -> list[str]:
+    return [
+        "af_alloy", "af_aoede", "af_bella", "af_heart", "af_jessica", "af_kore", "af_nicole", "af_nova", "af_river", "af_sarah", "af_sky",
+        "am_adam", "am_echo", "am_eric", "am_fenrir", "am_liam", "am_michael", "am_onyx", "am_puck", "am_santa",
+        "bf_alice", "bf_emma", "bf_isabella", "bf_lily",
+        "bm_daniel", "bm_fable", "bm_george", "bm_lewis",
+        "ef_dora", "em_alex", "em_santa",
+        "ff_siwis",
+        "hf_alpha", "hf_beta", "hm_omega", "hm_psi",
+        "if_sara", "im_nicola",
+        "jf_alpha", "jf_gongitsune", "jf_nezumi", "jf_tebukuro", "jm_kumo",
+        "pf_dora", "pm_alex", "pm_santa",
+        "zf_xiaobei", "zf_xiaoni", "zf_xiaoxiao", "zf_xiaoyi",
+        "zm_yunjian", "zm_yunxi", "zm_yunxia", "zm_yunyang",
+    ]
+
+
+def _init_kokoro() -> None:
+    if _KOKORO_STATE["initialized"]:
+        return
+    if KModel is None or KPipeline is None:
+        raise RuntimeError("Kokoro is not installed. Please install the 'kokoro' package (>=0.9.4).")
+    device = "cpu"
+    if torch is not None:
+        try:
+            if torch.cuda.is_available():
+                device = "cuda"
+        except Exception:
+            device = "cpu"
+    model = KModel().to(device).eval()
+    pipelines = {"a": KPipeline(lang_code="a", model=False)}
+    try:
+        pipelines["a"].g2p.lexicon.golds["kokoro"] = "kˈOkəɹO"
+    except Exception:
+        pass
+    _KOKORO_STATE.update({"initialized": True, "device": device, "model": model, "pipelines": pipelines})
+
+
+def List_Kokoro_Voices() -> list[str]:
+    return get_kokoro_voices()
+
+
+def Generate_Speech(
+    text: Annotated[str, "The text to synthesize (English)."],
+    speed: Annotated[float, "Speech speed multiplier in 0.5–2.0; 1.0 = normal speed."] = 1.25,
+    voice: Annotated[str, "Voice identifier from 54 available options."] = "af_heart",
+) -> tuple[int, np.ndarray]:
+    _log_call_start("Generate_Speech", text=_truncate_for_log(text, 200), speed=speed, voice=voice)
+    if not text or not text.strip():
+        try:
+            _log_call_end("Generate_Speech", "error=empty text")
+        finally:
+            pass
+        raise gr.Error("Please provide non-empty text to synthesize.")
+    _init_kokoro()
+    model = _KOKORO_STATE["model"]
+    pipelines = _KOKORO_STATE["pipelines"]
+    pipeline = pipelines.get("a")
+    if pipeline is None:
+        raise gr.Error("Kokoro English pipeline not initialized.")
+    audio_segments = []
+    pack = pipeline.load_voice(voice)
+    try:
+        segments = list(pipeline(text, voice, speed))
+        total_segments = len(segments)
+        for segment_idx, (text_chunk, ps, _) in enumerate(segments):
+            ref_s = pack[len(ps) - 1]
+            try:
+                audio = model(ps, ref_s, float(speed))
+                audio_segments.append(audio.detach().cpu().numpy())
+                if total_segments > 10 and (segment_idx + 1) % 5 == 0:
+                    print(f"Progress: Generated {segment_idx + 1}/{total_segments} segments...")
+            except Exception as exc:
+                raise gr.Error(f"Error generating audio for segment {segment_idx + 1}: {exc}")
+        if not audio_segments:
+            raise gr.Error("No audio was generated (empty synthesis result).")
+        if len(audio_segments) == 1:
+            final_audio = audio_segments[0]
+        else:
+            final_audio = np.concatenate(audio_segments, axis=0)
+            if total_segments > 1:
+                duration = len(final_audio) / 24_000
+                print(f"Completed: {total_segments} segments concatenated into {duration:.1f} seconds of audio")
+        _log_call_end("Generate_Speech", f"samples={final_audio.shape[0]} duration_sec={len(final_audio)/24_000:.2f}")
+        return 24_000, final_audio
+    except gr.Error as exc:
+        _log_call_end("Generate_Speech", f"gr_error={str(exc)}")
+        raise
+    except Exception as exc:  # pylint: disable=broad-except
+        _log_call_end("Generate_Speech", f"error={str(exc)[:120]}")
+        raise gr.Error(f"Error during speech generation: {exc}")
+
+
+def build_interface() -> gr.Interface:
+    available_voices = get_kokoro_voices()
+    return gr.Interface(
+        fn=Generate_Speech,
+        inputs=[
+            gr.Textbox(label="Text", placeholder="Type text to synthesize…", lines=4),
+            gr.Slider(minimum=0.5, maximum=2.0, value=1.25, step=0.1, label="Speed"),
+            gr.Dropdown(
+                label="Voice",
+                choices=available_voices,
+                value="af_heart",
+                info="Select from 54 available voices across multiple languages and accents",
+            ),
+        ],
+        outputs=gr.Audio(label="Audio", type="numpy", format="wav", show_download_button=True),
+        title="Generate Speech",
+        description=(
+            "<div style=\"text-align:center\">Generate speech with Kokoro-82M. Supports multiple languages and accents. Runs on CPU or CUDA if available.</div>"
+        ),
+        api_description=(
+            "Synthesize speech from text using Kokoro-82M TTS model. Returns (sample_rate, waveform) suitable for playback. "
+            "Parameters: text (str), speed (float 0.5–2.0, default 1.25x), voice (str, default 'af_heart'). "
+            "Voice Legend: af=American female, am=American male, bf=British female, bm=British male, ef=European female, em=European male, hf=Hindi female, hm=Hindi male, if=Italian female, im=Italian male, jf=Japanese female, jm=Japanese male, pf=Portuguese female, pm=Portuguese male, zf=Chinese female, zm=Chinese male, ff=French female. "
+            "All Voices: af_alloy, af_aoede, af_bella, af_heart, af_jessica, af_kore, af_nicole, af_nova, af_river, af_sarah, af_sky, am_adam, am_echo, am_eric, am_fenrir, am_liam, am_michael, am_onyx, am_puck, am_santa, bf_alice, bf_emma, bf_isabella, bf_lily, bm_daniel, bm_fable, bm_george, bm_lewis, ef_dora, em_alex, em_santa, ff_siwis, hf_alpha, hf_beta, hm_omega, hm_psi, if_sara, im_nicola, jf_alpha, jf_gongitsune, jf_nezumi, jf_tebukuro, jm_kumo, pf_dora, pm_alex, pm_santa, zf_xiaobei, zf_xiaoni, zf_xiaoxiao, zf_xiaoyi, zm_yunjian, zm_yunxi, zm_yunxia, zm_yunyang. "
+            "Return the generated media to the user in this format `![Alt text](URL)`"
+        ),
+        flagging_mode="never",
+    )
+
+
+__all__ = ["Generate_Speech", "List_Kokoro_Voices", "build_interface"]

Modules/Generate_Video.py

@@ -0,0 +1,171 @@
+from __future__ import annotations
+
+import os
+import random
+import tempfile
+from typing import Annotated
+
+import gradio as gr
+from huggingface_hub import InferenceClient
+
+from app import _log_call_end, _log_call_start, _truncate_for_log
+
+HF_VIDEO_TOKEN = os.getenv("HF_READ_TOKEN") or os.getenv("HF_TOKEN")
+
+
+def _write_video_tmp(data_iter_or_bytes: object, suffix: str = ".mp4") -> str:
+    fd, fname = tempfile.mkstemp(suffix=suffix)
+    try:
+        with os.fdopen(fd, "wb") as file:
+            if isinstance(data_iter_or_bytes, (bytes, bytearray)):
+                file.write(data_iter_or_bytes)
+            elif hasattr(data_iter_or_bytes, "read"):
+                file.write(data_iter_or_bytes.read())
+            elif hasattr(data_iter_or_bytes, "content"):
+                file.write(data_iter_or_bytes.content)  # type: ignore[attr-defined]
+            elif hasattr(data_iter_or_bytes, "__iter__") and not isinstance(data_iter_or_bytes, (str, dict)):
+                for chunk in data_iter_or_bytes:  # type: ignore[assignment]
+                    if chunk:
+                        file.write(chunk)
+            else:
+                raise gr.Error("Unsupported video data type returned by provider.")
+    except Exception:
+        try:
+            os.remove(fname)
+        except Exception:
+            pass
+        raise
+    return fname
+
+
+def Generate_Video(
+    prompt: Annotated[str, "Text description of the video to generate (e.g., 'a red fox running through a snowy forest at sunrise')."],
+    model_id: Annotated[str, "Hugging Face model id in the form 'creator/model-name'. Defaults to Wan-AI/Wan2.2-T2V-A14B."] = "Wan-AI/Wan2.2-T2V-A14B",
+    negative_prompt: Annotated[str, "What should NOT appear in the video."] = "",
+    steps: Annotated[int, "Number of denoising steps (1–100). Higher can improve quality but is slower."] = 25,
+    cfg_scale: Annotated[float, "Guidance scale (1–20). Higher = follow the prompt more closely, lower = more creative."] = 3.5,
+    seed: Annotated[int, "Random seed for reproducibility. Use -1 for a random seed per call."] = -1,
+    width: Annotated[int, "Output width in pixels (multiples of 8 recommended)."] = 768,
+    height: Annotated[int, "Output height in pixels (multiples of 8 recommended)."] = 768,
+    fps: Annotated[int, "Frames per second of the output video (e.g., 24)."] = 24,
+    duration: Annotated[float, "Target duration in seconds (provider/model dependent, commonly 2–6s)."] = 4.0,
+) -> str:
+    _log_call_start(
+        "Generate_Video",
+        prompt=_truncate_for_log(prompt, 160),
+        model_id=model_id,
+        steps=steps,
+        cfg_scale=cfg_scale,
+        fps=fps,
+        duration=duration,
+        size=f"{width}x{height}",
+    )
+    if not prompt or not prompt.strip():
+        _log_call_end("Generate_Video", "error=empty prompt")
+        raise gr.Error("Please provide a non-empty prompt.")
+    providers = ["auto", "replicate", "fal-ai"]
+    last_error: Exception | None = None
+    parameters = {
+        "negative_prompt": negative_prompt or None,
+        "num_inference_steps": steps,
+        "guidance_scale": cfg_scale,
+        "seed": seed if seed != -1 else random.randint(1, 1_000_000_000),
+        "width": width,
+        "height": height,
+        "fps": fps,
+        "duration": duration,
+    }
+    for provider in providers:
+        try:
+            client = InferenceClient(api_key=HF_VIDEO_TOKEN, provider=provider)
+            if hasattr(client, "text_to_video"):
+                num_frames = int(duration * fps) if duration and fps else None
+                extra_body = {}
+                if width:
+                    extra_body["width"] = width
+                if height:
+                    extra_body["height"] = height
+                if fps:
+                    extra_body["fps"] = fps
+                if duration:
+                    extra_body["duration"] = duration
+                result = client.text_to_video(
+                    prompt=prompt,
+                    model=model_id,
+                    guidance_scale=cfg_scale,
+                    negative_prompt=[negative_prompt] if negative_prompt else None,
+                    num_frames=num_frames,
+                    num_inference_steps=steps,
+                    seed=parameters["seed"],
+                    extra_body=extra_body if extra_body else None,
+                )
+            else:
+                result = client.post(
+                    model=model_id,
+                    json={"inputs": prompt, "parameters": {k: v for k, v in parameters.items() if v is not None}},
+                )
+            path = _write_video_tmp(result, suffix=".mp4")
+            try:
+                size = os.path.getsize(path)
+            except Exception:
+                size = -1
+            _log_call_end("Generate_Video", f"provider={provider} path={os.path.basename(path)} bytes={size}")
+            return path
+        except Exception as exc:  # pylint: disable=broad-except
+            last_error = exc
+            continue
+    msg = str(last_error) if last_error else "Unknown error"
+    lowered = msg.lower()
+    if "404" in msg:
+        raise gr.Error(f"Model not found or unavailable: {model_id}. Check the id and HF token access.")
+    if "503" in msg:
+        raise gr.Error("The model is warming up. Please try again shortly.")
+    if "401" in msg or "403" in msg:
+        raise gr.Error("Please duplicate the space and provide a `HF_READ_TOKEN` to enable Image and Video Generation.")
+    if ("api_key" in lowered) or ("hf auth login" in lowered) or ("unauthorized" in lowered) or ("forbidden" in lowered):
+        raise gr.Error("Please duplicate the space and provide a `HF_READ_TOKEN` to enable Image and Video Generation.")
+    _log_call_end("Generate_Video", f"error={_truncate_for_log(msg, 200)}")
+    raise gr.Error(f"Video generation failed: {msg}")
+
+
+def build_interface() -> gr.Interface:
+    return gr.Interface(
+        fn=Generate_Video,
+        inputs=[
+            gr.Textbox(label="Prompt", placeholder="Enter a prompt for the video", lines=2),
+            gr.Textbox(
+                label="Model",
+                value="Wan-AI/Wan2.2-T2V-A14B",
+                placeholder="creator/model-name",
+                max_lines=1,
+                info="<a href=\"https://huggingface.co/models?pipeline_tag=text-to-video&inference_provider=nebius,cerebras,novita,fireworks-ai,together,fal-ai,groq,featherless-ai,nscale,hyperbolic,sambanova,cohere,replicate,scaleway,publicai,hf-inference&sort=trending\" target=\"_blank\" rel=\"noopener noreferrer\">Browse models</a>",
+            ),
+            gr.Textbox(label="Negative Prompt", value="", lines=2),
+            gr.Slider(minimum=1, maximum=100, value=25, step=1, label="Steps"),
+            gr.Slider(minimum=1.0, maximum=20.0, value=3.5, step=0.1, label="CFG Scale"),
+            gr.Slider(minimum=-1, maximum=1_000_000_000, value=-1, step=1, label="Seed (-1 = random)"),
+            gr.Slider(minimum=64, maximum=1920, value=768, step=8, label="Width"),
+            gr.Slider(minimum=64, maximum=1920, value=768, step=8, label="Height"),
+            gr.Slider(minimum=4, maximum=60, value=24, step=1, label="FPS"),
+            gr.Slider(minimum=1.0, maximum=10.0, value=4.0, step=0.5, label="Duration (s)"),
+        ],
+        outputs=gr.Video(label="Generated Video", show_download_button=True, format="mp4"),
+        title="Generate Video",
+        description=(
+            "<div style=\"text-align:center\">Generate short videos via Hugging Face serverless inference. "
+            "Default model is Wan2.2-T2V-A14B.</div>"
+        ),
+        api_description=(
+            "Generate a short video from a text prompt using a Hugging Face model via serverless inference. "
+            "Create dynamic scenes like 'a red fox running through a snowy forest at sunrise', 'waves crashing on a rocky shore', "
+            "'time-lapse of clouds moving across a blue sky'. Default model: Wan2.2-T2V-A14B (2-6 second videos). "
+            "Parameters: prompt (str), model_id (str), negative_prompt (str), steps (int), cfg_scale (float), seed (int), "
+            "width/height (int), fps (int), duration (float in seconds). Returns MP4 file path. "
+            "Return the generated media to the user in this format `![Alt text](URL)`"
+        ),
+        flagging_mode="never",
+        show_api=bool(os.getenv("HF_READ_TOKEN") or os.getenv("HF_TOKEN")),
+    )
+
+
+__all__ = ["Generate_Video", "build_interface"]

Modules/Memory_Manager.py

@@ -0,0 +1,247 @@
+from __future__ import annotations
+
+import json
+import os
+import threading
+import uuid
+from datetime import datetime
+from typing import Annotated, Dict, List, Literal, Optional
+
+import gradio as gr
+
+MEMORY_FILE = os.path.join(os.path.dirname(__file__), "memories.json")
+_MEMORY_LOCK = threading.RLock()
+_MAX_MEMORIES = 10_000
+
+
+def _now_iso() -> str:
+    return datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
+
+
+def _load_memories() -> List[Dict[str, str]]:
+    if not os.path.exists(MEMORY_FILE):
+        return []
+    try:
+        with open(MEMORY_FILE, "r", encoding="utf-8") as file:
+            data = json.load(file)
+        if isinstance(data, list):
+            cleaned: List[Dict[str, str]] = []
+            for item in data:
+                if isinstance(item, dict) and "id" in item and "text" in item:
+                    cleaned.append(item)
+            return cleaned
+        return []
+    except Exception:
+        try:
+            backup = MEMORY_FILE + ".corrupt"
+            if not os.path.exists(backup):
+                os.replace(MEMORY_FILE, backup)
+        except Exception:
+            pass
+        return []
+
+
+def _save_memories(memories: List[Dict[str, str]]) -> None:
+    tmp_path = MEMORY_FILE + ".tmp"
+    with open(tmp_path, "w", encoding="utf-8") as file:
+        json.dump(memories, file, ensure_ascii=False, indent=2)
+    os.replace(tmp_path, MEMORY_FILE)
+
+
+def _mem_save(text: str, tags: str) -> str:
+    text_clean = (text or "").strip()
+    if not text_clean:
+        return "Error: memory text is empty."
+    with _MEMORY_LOCK:
+        memories = _load_memories()
+        if memories and memories[-1].get("text") == text_clean:
+            return "Skipped: identical to last stored memory."
+        mem_id = str(uuid.uuid4())
+        entry = {
+            "id": mem_id,
+            "text": text_clean,
+            "timestamp": _now_iso(),
+            "tags": tags.strip(),
+        }
+        memories.append(entry)
+        if len(memories) > _MAX_MEMORIES:
+            overflow = len(memories) - _MAX_MEMORIES
+            memories = memories[overflow:]
+        _save_memories(memories)
+    return f"Memory saved: {mem_id}"
+
+
+def _mem_list(limit: int, include_tags: bool) -> str:
+    limit = max(1, min(200, limit))
+    with _MEMORY_LOCK:
+        memories = _load_memories()
+    if not memories:
+        return "No memories stored yet."
+    chosen = memories[-limit:][::-1]
+    lines: List[str] = []
+    for memory in chosen:
+        base = f"{memory['id'][:8]} [{memory.get('timestamp','?')}] {memory.get('text','')}"
+        if include_tags and memory.get("tags"):
+            base += f" | tags: {memory['tags']}"
+        lines.append(base)
+    omitted = len(memories) - len(chosen)
+    if omitted > 0:
+        lines.append(f"… ({omitted} older memorie{'s' if omitted!=1 else ''} omitted; total={len(memories)})")
+    return "\n".join(lines)
+
+
+def _parse_search_query(query: str) -> Dict[str, List[str]]:
+    import re
+
+    result = {"tag_terms": [], "text_terms": [], "operator": "and"}
+    if not query or not query.strip():
+        return result
+    query = re.sub(r"\s+", " ", query.strip())
+    if re.search(r"\bOR\b", query, re.IGNORECASE):
+        result["operator"] = "or"
+        parts = re.split(r"\s+OR\s+", query, flags=re.IGNORECASE)
+    else:
+        parts = re.split(r"\s+(?:AND\s+)?", query, flags=re.IGNORECASE)
+        parts = [p for p in parts if p.strip() and p.strip().upper() != "AND"]
+    for part in parts:
+        part = part.strip()
+        if not part:
+            continue
+        tag_match = re.match(r"^tag:(.+)$", part, re.IGNORECASE)
+        if tag_match:
+            tag_name = tag_match.group(1).strip()
+            if tag_name:
+                result["tag_terms"].append(tag_name.lower())
+        else:
+            result["text_terms"].append(part.lower())
+    return result
+
+
+def _match_memory_with_query(memory: Dict[str, str], parsed_query: Dict[str, List[str]]) -> bool:
+    tag_terms = parsed_query["tag_terms"]
+    text_terms = parsed_query["text_terms"]
+    operator = parsed_query["operator"]
+    if not tag_terms and not text_terms:
+        return False
+    memory_text = memory.get("text", "").lower()
+    memory_tags = memory.get("tags", "").lower()
+    memory_tag_list = [tag.strip() for tag in memory_tags.split(",") if tag.strip()]
+    tag_matches = [any(tag_term in tag for tag in memory_tag_list) for tag_term in tag_terms]
+    combined_text = memory_text + " " + memory_tags
+    text_matches = [text_term in combined_text for text_term in text_terms]
+    all_matches = tag_matches + text_matches
+    if not all_matches:
+        return False
+    if operator == "or":
+        return any(all_matches)
+    return all(all_matches)
+
+
+def _mem_search(query: str, limit: int) -> str:
+    q = (query or "").strip()
+    if not q:
+        return "Error: empty query."
+    parsed_query = _parse_search_query(q)
+    if not parsed_query["tag_terms"] and not parsed_query["text_terms"]:
+        return "Error: no valid search terms found."
+    limit = max(1, min(200, limit))
+    with _MEMORY_LOCK:
+        memories = _load_memories()
+    matches: List[Dict[str, str]] = []
+    total_matches = 0
+    for memory in reversed(memories):
+        if _match_memory_with_query(memory, parsed_query):
+            total_matches += 1
+            if len(matches) < limit:
+                matches.append(memory)
+    if not matches:
+        return f"No matches for: {query}"
+    lines = [
+        f"{memory['id'][:8]} [{memory.get('timestamp','?')}] {memory.get('text','')}" + (f" | tags: {memory['tags']}" if memory.get('tags') else "")
+        for memory in matches
+    ]
+    omitted = total_matches - len(matches)
+    if omitted > 0:
+        lines.append(f"… ({omitted} additional match{'es' if omitted!=1 else ''} omitted; total_matches={total_matches})")
+    return "\n".join(lines)
+
+
+def _mem_delete(memory_id: str) -> str:
+    key = (memory_id or "").strip().lower()
+    if len(key) < 4:
+        return "Error: supply at least 4 characters of the id."
+    with _MEMORY_LOCK:
+        memories = _load_memories()
+        matched = [memory for memory in memories if memory["id"].lower().startswith(key)]
+        if not matched:
+            return "Memory not found."
+        if len(matched) > 1 and key != matched[0]["id"].lower():
+            sample = ", ".join(memory["id"][:8] for memory in matched[:5])
+            more = "…" if len(matched) > 5 else ""
+            return f"Ambiguous prefix (matches {len(matched)} ids: {sample}{more}). Provide more characters."
+        target_id = matched[0]["id"]
+        memories = [memory for memory in memories if memory["id"] != target_id]
+        _save_memories(memories)
+    return f"Deleted memory: {target_id}"
+
+
+def Memory_Manager(
+    action: Annotated[Literal["save", "list", "search", "delete"], "Action to perform: save | list | search | delete"],
+    text: Annotated[Optional[str], "Text content (Save only)"] = None,
+    tags: Annotated[Optional[str], "Comma-separated tags (Save only)"] = None,
+    query: Annotated[Optional[str], "Enhanced search with tag:name syntax, AND/OR operators (Search only)"] = None,
+    limit: Annotated[int, "Max results (List/Search only)"] = 20,
+    memory_id: Annotated[Optional[str], "Full UUID or unique prefix (Delete only)"] = None,
+    include_tags: Annotated[bool, "Include tags (List/Search only)"] = True,
+) -> str:
+    act = (action or "").lower().strip()
+    text = text or ""
+    tags = tags or ""
+    query = query or ""
+    memory_id = memory_id or ""
+    if act == "save":
+        if not text.strip():
+            return "Error: 'text' is required when action=save."
+        return _mem_save(text=text, tags=tags)
+    if act == "list":
+        return _mem_list(limit=limit, include_tags=include_tags)
+    if act == "search":
+        if not query.strip():
+            return "Error: 'query' is required when action=search."
+        return _mem_search(query=query, limit=limit)
+    if act == "delete":
+        if not memory_id.strip():
+            return "Error: 'memory_id' is required when action=delete."
+        return _mem_delete(memory_id=memory_id)
+    return "Error: invalid action (use save|list|search|delete)."
+
+
+def build_interface() -> gr.Interface:
+    return gr.Interface(
+        fn=Memory_Manager,
+        inputs=[
+            gr.Dropdown(label="Action", choices=["save", "list", "search", "delete"], value="list"),
+            gr.Textbox(label="Text", lines=3, placeholder="Memory text (save)"),
+            gr.Textbox(label="Tags", placeholder="tag1, tag2", max_lines=1),
+            gr.Textbox(label="Query", placeholder="tag:work AND tag:project OR meeting", max_lines=1),
+            gr.Slider(1, 200, value=20, step=1, label="Limit"),
+            gr.Textbox(label="Memory ID / Prefix", placeholder="UUID or prefix (delete)", max_lines=1),
+            gr.Checkbox(value=True, label="Include Tags"),
+        ],
+        outputs=gr.Textbox(label="Result", lines=14),
+        title="Memory Manager",
+        description=(
+            "<div style=\"text-align:center\">Lightweight local JSON memory store (no external DB). Choose an Action, fill only the relevant fields, and run.</div>"
+        ),
+        api_description=(
+            "Manage short text memories with optional tags. Actions: save(text,tags), list(limit,include_tags), "
+            "search(query,limit,include_tags), delete(memory_id). Enhanced search supports tag:name queries and AND/OR operators. "
+            "Examples: 'tag:work', 'tag:work AND tag:project', 'meeting tag:work', 'tag:urgent OR important'. "
+            "Action parameter is always required. Use Memory_Manager whenever you are given information worth remembering about the user, "
+            "and search for memories when relevant."
+        ),
+        flagging_mode="never",
+    )
+
+
+__all__ = ["Memory_Manager", "build_interface", "_load_memories", "_save_memories"]

Modules/Web_Fetch.py

@@ -0,0 +1,280 @@
+from __future__ import annotations
+
+import re
+from typing import Annotated, Dict, List, Tuple
+from urllib.parse import urlparse, urljoin
+
+import gradio as gr
+import requests
+from bs4 import BeautifulSoup
+from markdownify import markdownify as md
+from readability import Document
+
+from app import _fetch_rate_limiter, _log_call_end, _log_call_start, _truncate_for_log
+
+
+def _http_get_enhanced(url: str, timeout: int | float = 30, *, skip_rate_limit: bool = False) -> requests.Response:
+    headers = {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
+        "Accept-Language": "en-US,en;q=0.9",
+        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
+        "Accept-Encoding": "gzip, deflate, br",
+        "DNT": "1",
+        "Connection": "keep-alive",
+        "Upgrade-Insecure-Requests": "1",
+    }
+    if not skip_rate_limit:
+        _fetch_rate_limiter.acquire()
+    try:
+        response = requests.get(
+            url,
+            headers=headers,
+            timeout=timeout,
+            allow_redirects=True,
+            stream=False,
+        )
+        response.raise_for_status()
+        return response
+    except requests.exceptions.Timeout as exc:
+        raise requests.exceptions.RequestException("Request timed out. The webpage took too long to respond.") from exc
+    except requests.exceptions.ConnectionError as exc:
+        raise requests.exceptions.RequestException("Connection error. Please check the URL and your internet connection.") from exc
+    except requests.exceptions.HTTPError as exc:
+        if response.status_code == 403:
+            raise requests.exceptions.RequestException("Access forbidden. The website may be blocking automated requests.") from exc
+        if response.status_code == 404:
+            raise requests.exceptions.RequestException("Page not found. Please check the URL.") from exc
+        if response.status_code == 429:
+            raise requests.exceptions.RequestException("Rate limited. Please try again in a few minutes.") from exc
+        raise requests.exceptions.RequestException(f"HTTP error {response.status_code}: {exc}") from exc
+
+
+def _normalize_whitespace(text: str) -> str:
+    text = re.sub(r"[ \t\u00A0]+", " ", text)
+    text = re.sub(r"\n\s*\n\s*\n+", "\n\n", text.strip())
+    return text.strip()
+
+
+def _truncate(text: str, max_chars: int) -> Tuple[str, bool]:
+    if max_chars is None or max_chars <= 0 or len(text) <= max_chars:
+        return text, False
+    return text[:max_chars].rstrip() + " …", True
+
+
+def _shorten(text: str, limit: int) -> str:
+    if limit <= 0 or len(text) <= limit:
+        return text
+    return text[: max(0, limit - 1)].rstrip() + "…"
+
+
+def _domain_of(url: str) -> str:
+    try:
+        return urlparse(url).netloc or ""
+    except Exception:
+        return ""
+
+
+def _extract_links_from_soup(soup: BeautifulSoup, base_url: str) -> str:
+    links = []
+    for link in soup.find_all("a", href=True):
+        href = link.get("href")
+        text = link.get_text(strip=True)
+        if href.startswith("http"):
+            full_url = href
+        elif href.startswith("//"):
+            full_url = "https:" + href
+        elif href.startswith("/"):
+            full_url = urljoin(base_url, href)
+        else:
+            full_url = urljoin(base_url, href)
+        if text and href not in ["#", "javascript:void(0)"]:
+            links.append(f"- [{text}]({full_url})")
+    if not links:
+        return "No links found on this page."
+    title = soup.find("title")
+    title_text = title.get_text(strip=True) if title else "Links from webpage"
+    return f"# {title_text}\n\n" + "\n".join(links)
+
+
+def _fullpage_markdown_from_soup(full_soup: BeautifulSoup, base_url: str, strip_selectors: str = "") -> str:
+    if strip_selectors:
+        selectors = [s.strip() for s in strip_selectors.split(",") if s.strip()]
+        for selector in selectors:
+            try:
+                for element in full_soup.select(selector):
+                    element.decompose()
+            except Exception:
+                continue
+    for element in full_soup.select("script, style, nav, footer, header, aside"):
+        element.decompose()
+    main = (
+        full_soup.find("main")
+        or full_soup.find("article")
+        or full_soup.find("div", class_=re.compile(r"content|main|post|article", re.I))
+        or full_soup.find("body")
+    )
+    if not main:
+        return "No main content found on the webpage."
+    markdown_text = md(str(main), heading_style="ATX")
+    markdown_text = re.sub(r"\n{3,}", "\n\n", markdown_text)
+    markdown_text = re.sub(r"\[\s*\]\([^)]*\)", "", markdown_text)
+    markdown_text = re.sub(r"[ \t]+", " ", markdown_text)
+    markdown_text = markdown_text.strip()
+    title = full_soup.find("title")
+    if title and title.get_text(strip=True):
+        markdown_text = f"# {title.get_text(strip=True)}\n\n{markdown_text}"
+    return markdown_text or "No content could be extracted."
+
+
+def _truncate_markdown(markdown: str, max_chars: int) -> Tuple[str, Dict[str, object]]:
+    total_chars = len(markdown)
+    if total_chars <= max_chars:
+        return markdown, {
+            "truncated": False,
+            "returned_chars": total_chars,
+            "total_chars_estimate": total_chars,
+            "next_cursor": None,
+        }
+    truncated = markdown[:max_chars]
+    last_paragraph = truncated.rfind("\n\n")
+    if last_paragraph > max_chars * 0.7:
+        truncated = truncated[:last_paragraph]
+        cursor_pos = last_paragraph
+    elif "." in truncated[-100:]:
+        last_period = truncated.rfind(".")
+        if last_period > max_chars * 0.8:
+            truncated = truncated[: last_period + 1]
+            cursor_pos = last_period + 1
+        else:
+            cursor_pos = len(truncated)
+    else:
+        cursor_pos = len(truncated)
+    metadata = {
+        "truncated": True,
+        "returned_chars": len(truncated),
+        "total_chars_estimate": total_chars,
+        "next_cursor": cursor_pos,
+    }
+    truncated = truncated.rstrip()
+    truncation_notice = (
+        "\n\n---\n"
+        f"**Content Truncated:** Showing {metadata['returned_chars']:,} of {metadata['total_chars_estimate']:,} characters "
+        f"({(metadata['returned_chars']/metadata['total_chars_estimate']*100):.1f}%)\n"
+        f"**Next cursor:** {metadata['next_cursor']} (use this value with offset parameter for continuation)\n"
+        "---"
+    )
+    return truncated + truncation_notice, metadata
+
+
+def Web_Fetch(
+    url: Annotated[str, "The absolute URL to fetch (must return HTML)."],
+    max_chars: Annotated[int, "Maximum characters to return (0 = no limit, full page content)."] = 3000,
+    strip_selectors: Annotated[str, "CSS selectors to remove (comma-separated, e.g., '.header, .footer, nav')."] = "",
+    url_scraper: Annotated[bool, "Extract only links from the page instead of content."] = False,
+    offset: Annotated[int, "Character offset to start from (for pagination, use next_cursor from previous call)."] = 0,
+) -> str:
+    _log_call_start(
+        "Web_Fetch",
+        url=url,
+        max_chars=max_chars,
+        strip_selectors=strip_selectors,
+        url_scraper=url_scraper,
+        offset=offset,
+    )
+    if not url or not url.strip():
+        result = "Please enter a valid URL."
+        _log_call_end("Web_Fetch", _truncate_for_log(result))
+        return result
+    try:
+        resp = _http_get_enhanced(url)
+        resp.raise_for_status()
+    except requests.exceptions.RequestException as exc:
+        result = f"An error occurred: {exc}"
+        _log_call_end("Web_Fetch", _truncate_for_log(result))
+        return result
+    final_url = str(resp.url)
+    ctype = resp.headers.get("Content-Type", "")
+    if "html" not in ctype.lower():
+        result = f"Unsupported content type for extraction: {ctype or 'unknown'}"
+        _log_call_end("Web_Fetch", _truncate_for_log(result))
+        return result
+    resp.encoding = resp.encoding or resp.apparent_encoding
+    html = resp.text
+    full_soup = BeautifulSoup(html, "lxml")
+    if url_scraper:
+        result = _extract_links_from_soup(full_soup, final_url)
+        if offset > 0:
+            result = result[offset:]
+        if max_chars > 0 and len(result) > max_chars:
+            result, _ = _truncate_markdown(result, max_chars)
+    else:
+        full_result = _fullpage_markdown_from_soup(full_soup, final_url, strip_selectors)
+        if offset > 0:
+            if offset >= len(full_result):
+                result = (
+                    f"Offset {offset} exceeds content length ({len(full_result)} characters). "
+                    f"Content ends at position {len(full_result)}."
+                )
+                _log_call_end("Web_Fetch", _truncate_for_log(result))
+                return result
+            result = full_result[offset:]
+        else:
+            result = full_result
+        if max_chars > 0 and len(result) > max_chars:
+            result, metadata = _truncate_markdown(result, max_chars)
+            if offset > 0:
+                metadata["total_chars_estimate"] = len(full_result)
+                metadata["next_cursor"] = offset + metadata["next_cursor"] if metadata["next_cursor"] else None
+    _log_call_end("Web_Fetch", f"chars={len(result)}, url_scraper={url_scraper}, offset={offset}")
+    return result
+
+
+def build_interface() -> gr.Interface:
+    return gr.Interface(
+        fn=Web_Fetch,
+        inputs=[
+            gr.Textbox(label="URL", placeholder="https://example.com/article", max_lines=1),
+            gr.Slider(minimum=0, maximum=20000, value=3000, step=100, label="Max Characters", info="0 = no limit (full page), default 3000"),
+            gr.Textbox(
+                label="Strip Selectors",
+                placeholder=".header, .footer, nav, .sidebar",
+                value="",
+                max_lines=1,
+                info="CSS selectors to remove (comma-separated)",
+            ),
+            gr.Checkbox(label="URL Scraper", value=False, info="Extract only links instead of content"),
+            gr.Slider(
+                minimum=0,
+                maximum=100000,
+                value=0,
+                step=100,
+                label="Offset",
+                info="Character offset to start from (use next_cursor from previous call for pagination)",
+            ),
+        ],
+        outputs=gr.Markdown(label="Extracted Content"),
+        title="Web Fetch",
+        description=(
+            "<div style=\"text-align:center\">Convert any webpage to clean Markdown format with precision controls, "
+            "or extract all links. Supports custom element removal, length limits, and pagination with offset.</div>"
+        ),
+        api_description=(
+            "Fetch a web page and return it converted to Markdown format or extract links with configurable options. "
+            "Includes enhanced truncation with detailed metadata and pagination support via offset parameter. "
+            "Parameters: url (str - absolute URL), max_chars (int - 0=no limit, default 3000), "
+            "strip_selectors (str - CSS selectors to remove, comma-separated), "
+            "url_scraper (bool - extract only links instead of content, default False), "
+            "offset (int - character offset for pagination, use next_cursor from previous call). "
+            "When content is truncated, returns detailed metadata including truncated status, character counts, "
+            "and next_cursor for continuation. When url_scraper=True, returns formatted list of all links found on the page."
+        ),
+        flagging_mode="never",
+    )
+
+
+__all__ = [
+    "Web_Fetch",
+    "build_interface",
+    "_http_get_enhanced",
+    "_fullpage_markdown_from_soup",
+]

Modules/Web_Search.py

@@ -0,0 +1,268 @@
+from __future__ import annotations
+
+from typing import Annotated, List
+
+import gradio as gr
+from ddgs import DDGS
+
+from app import _log_call_end, _log_call_start, _search_rate_limiter, _truncate_for_log
+
+
+def _extract_date_from_snippet(snippet: str) -> str:
+    if not snippet:
+        return ""
+    import re
+
+    date_patterns = [
+        r"\b(\d{4}[-/]\d{1,2}[-/]\d{1,2})\b",
+        r"\b([A-Za-z]{3,9}\s+\d{1,2},?\s+\d{4})\b",
+        r"\b(\d{1,2}\s+[A-Za-z]{3,9}\s+\d{4})\b",
+        r"\b(\d+\s+(?:day|week|month|year)s?\s+ago)\b",
+        r"(?:Published|Updated|Posted):\s*([^,\n]+?)(?:[,\n]|$)",
+    ]
+    for pattern in date_patterns:
+        matches = re.findall(pattern, snippet, re.IGNORECASE)
+        if matches:
+            return matches[0].strip()
+    return ""
+
+
+def _format_search_result(result: dict, search_type: str, index: int) -> List[str]:
+    lines: List[str] = []
+    if search_type == "text":
+        title = result.get("title", "").strip()
+        url = result.get("href", "").strip()
+        snippet = result.get("body", "").strip()
+        date = _extract_date_from_snippet(snippet)
+        lines.append(f"{index}. {title}")
+        lines.append(f"   URL: {url}")
+        if snippet:
+            lines.append(f"   Summary: {snippet}")
+        if date:
+            lines.append(f"   Date: {date}")
+    elif search_type == "news":
+        title = result.get("title", "").strip()
+        url = result.get("url", "").strip()
+        body = result.get("body", "").strip()
+        date = result.get("date", "").strip()
+        source = result.get("source", "").strip()
+        lines.append(f"{index}. {title}")
+        lines.append(f"   URL: {url}")
+        if source:
+            lines.append(f"   Source: {source}")
+        if date:
+            lines.append(f"   Date: {date}")
+        if body:
+            lines.append(f"   Summary: {body}")
+    elif search_type == "images":
+        title = result.get("title", "").strip()
+        image_url = result.get("image", "").strip()
+        source_url = result.get("url", "").strip()
+        source = result.get("source", "").strip()
+        width = result.get("width", "")
+        height = result.get("height", "")
+        lines.append(f"{index}. {title}")
+        lines.append(f"   Image: {image_url}")
+        lines.append(f"   Source: {source_url}")
+        if source:
+            lines.append(f"   Publisher: {source}")
+        if width and height:
+            lines.append(f"   Dimensions: {width}x{height}")
+    elif search_type == "videos":
+        title = result.get("title", "").strip()
+        description = result.get("description", "").strip()
+        duration = result.get("duration", "").strip()
+        published = result.get("published", "").strip()
+        uploader = result.get("uploader", "").strip()
+        embed_url = result.get("embed_url", "").strip()
+        lines.append(f"{index}. {title}")
+        if embed_url:
+            lines.append(f"   Video: {embed_url}")
+        if uploader:
+            lines.append(f"   Uploader: {uploader}")
+        if duration:
+            lines.append(f"   Duration: {duration}")
+        if published:
+            lines.append(f"   Published: {published}")
+        if description:
+            lines.append(f"   Description: {description}")
+    elif search_type == "books":
+        title = result.get("title", "").strip()
+        url = result.get("url", "").strip()
+        body = result.get("body", "").strip()
+        lines.append(f"{index}. {title}")
+        lines.append(f"   URL: {url}")
+        if body:
+            lines.append(f"   Description: {body}")
+    return lines
+
+
+def Web_Search(
+    query: Annotated[str, "The search query (supports operators like site:, quotes, OR)."],
+    max_results: Annotated[int, "Number of results to return (1–20)."] = 5,
+    page: Annotated[int, "Page number for pagination (1-based, each page contains max_results items)."] = 1,
+    search_type: Annotated[str, "Type of search: 'text' (web pages), 'news', 'images', 'videos', or 'books'."] = "text",
+    offset: Annotated[int, "Result offset to start from (overrides page if > 0, for precise continuation)."] = 0,
+) -> str:
+    _log_call_start("Web_Search", query=query, max_results=max_results, page=page, search_type=search_type, offset=offset)
+    if not query or not query.strip():
+        result = "No search query provided. Please enter a search term."
+        _log_call_end("Web_Search", _truncate_for_log(result))
+        return result
+    max_results = max(1, min(20, max_results))
+    page = max(1, page)
+    offset = max(0, offset)
+    valid_types = ["text", "news", "images", "videos", "books"]
+    if search_type not in valid_types:
+        search_type = "text"
+    if offset > 0:
+        actual_offset = offset
+        calculated_page = (offset // max_results) + 1
+    else:
+        actual_offset = (page - 1) * max_results
+        calculated_page = page
+    total_needed = actual_offset + max_results
+    used_fallback = False
+    original_search_type = search_type
+
+    def _perform_search(stype: str) -> list[dict]:
+        try:
+            _search_rate_limiter.acquire()
+            with DDGS() as ddgs:
+                if stype == "text":
+                    raw_gen = ddgs.text(query, max_results=total_needed + 10)
+                elif stype == "news":
+                    raw_gen = ddgs.news(query, max_results=total_needed + 10)
+                elif stype == "images":
+                    raw_gen = ddgs.images(query, max_results=total_needed + 10)
+                elif stype == "videos":
+                    raw_gen = ddgs.videos(query, max_results=total_needed + 10)
+                else:
+                    raw_gen = ddgs.books(query, max_results=total_needed + 10)
+                try:
+                    return list(raw_gen)
+                except Exception as inner_exc:
+                    if "no results" in str(inner_exc).lower() or "not found" in str(inner_exc).lower():
+                        return []
+                    raise inner_exc
+        except Exception as exc:
+            error_msg = f"Search failed: {str(exc)[:200]}"
+            lowered = str(exc).lower()
+            if "blocked" in lowered or "rate" in lowered:
+                error_msg = "Search temporarily blocked due to rate limiting. Please try again in a few minutes."
+            elif "timeout" in lowered:
+                error_msg = "Search timed out. Please try again with a simpler query."
+            elif "network" in lowered or "connection" in lowered:
+                error_msg = "Network connection error. Please check your internet connection and try again."
+            elif "no results" in lowered or "not found" in lowered:
+                return []
+            raise Exception(error_msg)
+
+    try:
+        raw = _perform_search(search_type)
+    except Exception as exc:
+        result = f"Error: {exc}"
+        _log_call_end("Web_Search", _truncate_for_log(result))
+        return result
+
+    if not raw and search_type == "news":
+        try:
+            raw = _perform_search("text")
+            if raw:
+                used_fallback = True
+                search_type = "text"
+        except Exception:
+            pass
+
+    if not raw:
+        fallback_note = " (also tried 'text' search as fallback)" if original_search_type == "news" and used_fallback else ""
+        result = f"No {original_search_type} results found for query: {query}{fallback_note}"
+        _log_call_end("Web_Search", _truncate_for_log(result))
+        return result
+
+    paginated_results = raw[actual_offset: actual_offset + max_results]
+    if not paginated_results:
+        if actual_offset >= len(raw):
+            result = f"Offset {actual_offset} exceeds available results ({len(raw)} total). Try offset=0 to start from beginning."
+        else:
+            result = f"No {original_search_type} results found on page {calculated_page} for query: {query}. Try page 1 or reduce page number."
+        _log_call_end("Web_Search", _truncate_for_log(result))
+        return result
+
+    total_available = len(raw)
+    start_num = actual_offset + 1
+    end_num = actual_offset + len(paginated_results)
+    next_offset = actual_offset + len(paginated_results)
+    search_label = original_search_type.title()
+    if used_fallback:
+        search_label += " → Text (Smart Fallback)"
+    pagination_info = f"Page {calculated_page}"
+    if offset > 0:
+        pagination_info = f"Offset {actual_offset} (≈ {pagination_info})"
+    lines = [f"{search_label} search results for: {query}"]
+    if used_fallback:
+        lines.append("📍 Note: News search returned no results, automatically searched general web content instead")
+    lines.append(f"{pagination_info} (results {start_num}-{end_num} of ~{total_available}+ available)\n")
+    for i, result in enumerate(paginated_results, start_num):
+        result_lines = _format_search_result(result, search_type, i)
+        lines.extend(result_lines)
+        lines.append("")
+    if total_available > end_num:
+        lines.append("💡 More results available:")
+        lines.append(f"   • Next page: page={calculated_page + 1}")
+        lines.append(f"   • Next offset: offset={next_offset}")
+        lines.append(f"   • Use offset={next_offset} to continue exactly from result {next_offset + 1}")
+    result = "\n".join(lines)
+    search_info = f"type={original_search_type}"
+    if used_fallback:
+        search_info += "→text"
+    _log_call_end("Web_Search", f"{search_info} page={calculated_page} offset={actual_offset} results={len(paginated_results)} chars={len(result)}")
+    return result
+
+
+def build_interface() -> gr.Interface:
+    return gr.Interface(
+        fn=Web_Search,
+        inputs=[
+            gr.Textbox(label="Query", placeholder="topic OR site:example.com", max_lines=1),
+            gr.Slider(minimum=1, maximum=20, value=5, step=1, label="Max results"),
+            gr.Slider(minimum=1, maximum=10, value=1, step=1, label="Page", info="Page number for pagination (ignored if offset > 0)"),
+            gr.Radio(
+                label="Search Type",
+                choices=["text", "news", "images", "videos", "books"],
+                value="text",
+                info="Type of content to search for",
+            ),
+            gr.Slider(
+                minimum=0,
+                maximum=1000,
+                value=0,
+                step=1,
+                label="Offset",
+                info="Result offset to start from (overrides page if > 0, use next_offset from previous search)",
+            ),
+        ],
+        outputs=gr.Textbox(label="Search Results", interactive=False, lines=20, max_lines=20),
+        title="Web Search",
+        description=(
+            "<div style=\"text-align:center\">Multi-type web search with readable output format, date detection, and flexible pagination. "
+            "Supports text, news, images, videos, and books. Features smart fallback for news searches and precise offset control.</div>"
+        ),
+        api_description=(
+            "Run a web search (DuckDuckGo backend) with support for multiple content types and return formatted results. "
+            "Features smart fallback: if 'news' search returns no results, automatically retries with 'text' search "
+            "to catch sources like Hacker News that might not appear in news-specific results. "
+            "Supports advanced search operators: site: for specific domains, quotes for exact phrases, "
+            "OR for alternatives, and - to exclude terms. Examples: 'Python programming', 'site:example.com', "
+            "'\"artificial intelligence\"', 'cats -dogs', 'Python OR JavaScript'. "
+            "Parameters: query (str), max_results (int, 1-20), page (int, 1-based pagination), "
+            "search_type (str: text/news/images/videos/books), offset (int, result offset for precise continuation). "
+            "If offset > 0, it overrides the page parameter. Returns appropriately formatted results with metadata, "
+            "pagination hints, and next_offset information for each content type."
+        ),
+        flagging_mode="never",
+        submit_btn="Search",
+    )
+
+
+__all__ = ["Web_Search", "build_interface"]

Modules/__init__.py

@@ -0,0 +1 @@
+"""Gradio tool modules bundled for Nymbo-Tools."""