Create search_pipeline.py

search_pipeline.py

@@ -0,0 +1,586 @@
+Infinity GRM - Search / Execution Pipeline
+===========================================
+
+Standalone tool/execution layer.
+
+Responsibilities:
+- conversation/session management
+- deciding when web research is needed
+- DuckDuckGo search
+- asynchronous web retrieval
+- page extraction
+- context assembly
+- calling the separate reasoning engine
+- FastAPI API
+
+The reasoning model itself lives in reasoning_engine.py.
+"""
+
+import os
+import re
+import asyncio
+import logging
+from contextlib import asynccontextmanager
+from urllib.parse import urlparse
+
+import httpx
+import torch
+import uvicorn
+from bs4 import BeautifulSoup
+from trafilatura import extract
+from ddgs import DDGS
+
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.staticfiles import StaticFiles
+from pydantic import BaseModel
+
+import reasoning_engine
+
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s",
+)
+
+logger = logging.getLogger("infinitygrm.pipeline")
+
+session_store = {}
+engine_state = {}
+
+
+# ---------------------------------------------------------------------
+# REQUEST MODEL
+# ---------------------------------------------------------------------
+
+class ChatRequest(BaseModel):
+    message: str
+    session_id: str
+
+
+# ---------------------------------------------------------------------
+# SESSION
+# ---------------------------------------------------------------------
+
+def fetch_or_create_session(session_id: str) -> dict:
+    if session_id not in session_store:
+        session_store[session_id] = {
+            "history": [],
+            "last_entity": "",
+            "last_web_query": "",
+            "last_response": "",
+            "last_web_context": [],
+        }
+
+    return session_store[session_id]
+
+
+def update_history(session: dict, role: str, content: str):
+    session["history"].append(
+        {
+            "role": role,
+            "content": content,
+        }
+    )
+
+    max_history = engine_state.get("max_history", 16)
+
+    while len(session["history"]) > max_history:
+        session["history"].pop(0)
+
+
+# ---------------------------------------------------------------------
+# FOLLOW-UP DETECTION
+# ---------------------------------------------------------------------
+
+def has_reference_language(query: str) -> bool:
+    words = set(re.findall(r"\b[\w'-]+\b", query.lower()))
+
+    return bool(
+        words.intersection(
+            {
+                "he", "him", "his",
+                "she", "her", "hers",
+                "they", "them", "their",
+                "it", "its",
+                "this", "that", "these", "those",
+                "then", "next", "after",
+            }
+        )
+    )
+
+
+def is_followup(query: str, session: dict) -> bool:
+    if not session.get("last_response"):
+        return False
+
+    if len(query.split()) <= 4:
+        return True
+
+    if has_reference_language(query):
+        return True
+
+    # Let the reasoning engine handle semantic interpretation when
+    # necessary rather than forcing a brittle keyword rule.
+    return False
+
+
+# ---------------------------------------------------------------------
+# WEB SEARCH
+# ---------------------------------------------------------------------
+
+def isolated_ddg_search(query: str, max_results: int = 8) -> list:
+    try:
+        with DDGS() as ddgs:
+            results = ddgs.text(
+                query,
+                max_results=max_results,
+            )
+
+            return [
+                {
+                    "title": r.get("title", ""),
+                    "href": r.get("href", ""),
+                    "body": r.get("body", ""),
+                }
+                for r in results
+                if r.get("href")
+            ]
+
+    except Exception as exc:
+        logger.error("DuckDuckGo search failed: %s", exc)
+        return []
+
+
+async def search_web_async(query: str) -> list:
+    return await asyncio.to_thread(
+        isolated_ddg_search,
+        query,
+    )
+
+
+# ---------------------------------------------------------------------
+# WEB SCRAPING
+# ---------------------------------------------------------------------
+
+async def scrape_target_url(client, result: dict):
+    url = result["href"]
+
+    try:
+        response = await client.get(
+            url,
+            timeout=12.0,
+            follow_redirects=True,
+        )
+
+        if response.status_code != 200:
+            logger.warning(
+                "HTTP %s: %s",
+                response.status_code,
+                url,
+            )
+            return None
+
+        text = None
+
+        try:
+            text = extract(response.text)
+        except Exception as exc:
+            logger.warning(
+                "Trafilatura failed for %s: %s",
+                url,
+                exc,
+            )
+
+        if not text:
+            try:
+                soup = BeautifulSoup(
+                    response.text,
+                    "html.parser",
+                )
+                text = soup.get_text(
+                    separator=" ",
+                    strip=True,
+                )
+            except Exception as exc:
+                logger.warning(
+                    "BeautifulSoup failed for %s: %s",
+                    url,
+                    exc,
+                )
+                return None
+
+        if not text or len(text.split()) < 20:
+            return None
+
+        return {
+            "url": url,
+            "title": result.get("title", ""),
+            "domain": urlparse(url).netloc,
+            "text": text[:7000],
+        }
+
+    except Exception as exc:
+        logger.warning(
+            "Scraping error for %s: %s",
+            url,
+            exc,
+        )
+        return None
+
+
+async def build_context_pipeline(
+    query: str,
+    max_results: int = 8,
+) -> tuple[str, list]:
+    """Search and scrape the web independently of reasoning."""
+    search_results = await search_web_async(
+        query
+    )
+
+    if not search_results:
+        return "", []
+
+    headers = {
+        "User-Agent": (
+            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+            "AppleWebKit/537.36 "
+            "(KHTML, like Gecko) "
+            "Chrome/122.0.0.0 Safari/537.36"
+        ),
+        "Accept": (
+            "text/html,application/xhtml+xml,"
+            "application/xml;q=0.9,image/avif,"
+            "image/webp,*/*;q=0.8"
+        ),
+        "Accept-Language": "en-US,en;q=0.9",
+        "Accept-Encoding": "gzip, deflate, br",
+        "Connection": "keep-alive",
+    }
+
+    try:
+        from curl_cffi.requests import AsyncSession
+    except ImportError:
+        AsyncSession = None
+
+    if AsyncSession is not None:
+        async with AsyncSession(
+            impersonate="chrome",
+            headers=headers,
+        ) as client:
+            tasks = [
+                scrape_target_url(client, result)
+                for result in search_results[:max_results]
+            ]
+
+            scraped = await asyncio.gather(
+                *tasks,
+                return_exceptions=False,
+            )
+    else:
+        async with httpx.AsyncClient(
+            headers=headers,
+            follow_redirects=True,
+        ) as client:
+            tasks = [
+                scrape_target_url(client, result)
+                for result in search_results[:max_results]
+            ]
+
+            scraped = await asyncio.gather(
+                *tasks,
+                return_exceptions=False,
+            )
+
+    blocks = []
+    sources = []
+
+    for item in scraped:
+        if not item:
+            continue
+
+        blocks.append(
+            "\n".join(
+                [
+                    f"SOURCE: {item['url']}",
+                    f"TITLE: {item['title']}",
+                    f"DOMAIN: {item['domain']}",
+                    f"CONTENT: {item['text']}",
+                ]
+            )
+        )
+
+        sources.append(item["url"])
+
+    return (
+        "\n\n---\n\n".join(blocks),
+        list(dict.fromkeys(sources)),
+    )
+
+
+# ---------------------------------------------------------------------
+# WEB DECISION
+# ---------------------------------------------------------------------
+
+SEARCH_TRIGGERS = {
+    "latest", "today", "current", "recent",
+    "news", "update", "price", "stock",
+    "founder", "ceo", "company",
+    "documentation", "release", "version",
+    "who", "what", "when", "where",
+    "code", "python", "javascript", "html",
+}
+
+
+def obvious_web_request(query: str) -> bool:
+    words = set(re.findall(r"\b[\w'-]+\b", query.lower()))
+    return bool(words.intersection(SEARCH_TRIGGERS))
+
+
+# ---------------------------------------------------------------------
+# MAIN EXECUTION
+# ---------------------------------------------------------------------
+
+async def process_request(
+    query: str,
+    session: dict,
+):
+    """
+    Separation boundary:
+
+        reasoning_engine.plan()
+                 |
+                 v
+        execution/search
+                 |
+                 v
+        reasoning_engine.analyze()
+                 |
+                 v
+        reasoning_engine.final_answer()
+    """
+
+    # ---------------------------------------------------------------
+    # PASS 1: REASONING / PLANNING
+    # ---------------------------------------------------------------
+
+    plan = await asyncio.to_thread(
+        reasoning_engine.plan,
+        query,
+        session["history"],
+    )
+
+    requires_web = bool(
+        plan.get("requires_web")
+        or obvious_web_request(query)
+        or is_followup(query, session)
+    )
+
+    search_queries = plan.get("search_queries") or [query]
+
+    # ---------------------------------------------------------------
+    # EXECUTION / SEARCH
+    # ---------------------------------------------------------------
+
+    evidence_blocks = []
+    sources = []
+
+    if requires_web:
+        # Multiple searches let the reasoning engine decompose advanced
+        # questions instead of forcing one broad query.
+        for search_query in search_queries[:4]:
+            search_query = str(search_query).strip()
+
+            if not search_query:
+                continue
+
+            context, found_sources = await build_context_pipeline(
+                search_query
+            )
+
+            if context:
+                evidence_blocks.append(
+                    f"SEARCH QUERY: {search_query}\n{context}"
+                )
+
+            sources.extend(found_sources)
+
+        sources = list(dict.fromkeys(sources))
+
+        session["last_web_query"] = " | ".join(
+            str(x) for x in search_queries[:4]
+        )
+
+        session["last_web_context"] = [
+            {
+                "user": query,
+                "search": session["last_web_query"],
+            }
+        ]
+
+    evidence = "\n\n========\n\n".join(
+        evidence_blocks
+    )
+
+    # ---------------------------------------------------------------
+    # PASS 2: REASONING / EVIDENCE ANALYSIS
+    # ---------------------------------------------------------------
+
+    analysis = await asyncio.to_thread(
+        reasoning_engine.analyze,
+        query,
+        plan,
+        evidence,
+        session["history"],
+    )
+
+    # ---------------------------------------------------------------
+    # PASS 3: FINAL SYNTHESIS
+    # ---------------------------------------------------------------
+
+    answer = await asyncio.to_thread(
+        reasoning_engine.final_answer,
+        query,
+        plan,
+        analysis,
+        evidence,
+        session["history"],
+    )
+
+    # ---------------------------------------------------------------
+    # LEARNING
+    # ---------------------------------------------------------------
+
+    # Store the solved problem as an experience. This does not mutate
+    # model weights; it creates a retrieval corpus for future problems.
+    await asyncio.to_thread(
+        reasoning_engine.learn_from_result,
+        query,
+        plan,
+        analysis,
+        answer,
+        1.0,
+    )
+
+    session["last_response"] = answer
+    session["last_entity"] = plan.get("task", "")
+
+    update_history(
+        session,
+        "user",
+        query,
+    )
+
+    update_history(
+        session,
+        "assistant",
+        answer,
+    )
+
+    return answer, sources
+
+
+# ---------------------------------------------------------------------
+# FASTAPI
+# ---------------------------------------------------------------------
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    logger.info("Starting Infinity GRM execution pipeline...")
+
+    await asyncio.to_thread(
+        reasoning_engine.initialize
+    )
+
+    engine_state["max_history"] = 16
+
+    yield
+
+    engine_state.clear()
+    logger.info("Infinity GRM pipeline stopped.")
+
+
+app = FastAPI(
+    title="Infinity GRM - Separated Reasoning/Search Engine",
+    lifespan=lifespan,
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.post("/query")
+async def chat_endpoint(request: ChatRequest):
+    try:
+        session = fetch_or_create_session(
+            request.session_id
+        )
+
+        answer, sources = await process_request(
+            request.message,
+            session,
+        )
+
+        return {
+            "response": answer,
+            "sources": sources,
+        }
+
+    except Exception as exc:
+        logger.exception(
+            "Critical exception in Infinity GRM pipeline"
+        )
+
+        raise HTTPException(
+            status_code=500,
+            detail=str(exc),
+        )
+
+
+@app.get("/health")
+async def health():
+    return {
+        "status": "healthy",
+        "gpu_acceleration_active": torch.cuda.is_available(),
+        "reasoning_engine": "separated",
+        "learning": "retrieval-based experience memory",
+    }
+
+
+if not os.path.exists("static"):
+    os.makedirs("static")
+
+app.mount(
+    "/static",
+    StaticFiles(directory="static"),
+    name="static",
+)
+
+
+@app.get("/")
+async def root():
+    return {
+        "message": "Infinity GRM separated reasoning/search engine is active.",
+        "architecture": [
+            "reasoning_engine",
+            "search_execution",
+            "evidence_analysis",
+            "final_synthesis",
+        ],
+    }
+
+
+if __name__ == "__main__":
+    port = int(os.getenv("PORT", "7860"))
+
+    uvicorn.run(
+        "search_pipeline:app",
+        host="0.0.0.0",
+        port=port,
+        reload=False,
+    )