Update search_pipeline.py

search_pipeline.py

@@ -1,588 +0,0 @@
-"""
-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,
-    )
-