#!/usr/bin/env python3 """ AI Research Engine — Backend API Runs on CT 145, handles all heavy lifting. Exposes REST API consumed by the MCP proxy (MacBook) and dashboard. """ import os import json import httpx import hashlib from fastapi import FastAPI, Query, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse import uvicorn app = FastAPI(title="AI Research Engine Backend") app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) # ── Config ────────────────────────────────────────────────── YACY_URL = os.getenv("YACY_URL", "http://localhost:8090") OPENSEARCH_URL = os.getenv("OPENSEARCH_URL", "http://localhost:9200") QDRANT_URL = os.getenv("QDRANT_URL", "http://10.30.20.68:6333") OLLAMA_URL = os.getenv("OLLAMA_URL", "http://10.30.20.186:11434") OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "ornith:latest") INDEX_NAME = os.getenv("INDEX_NAME", "research_docs") client = httpx.Client(timeout=30.0) ollama = httpx.Client(timeout=120.0, base_url=OLLAMA_URL) # ── Helpers ────────────────────────────────────────────────── def _ensure_index(): try: client.get(f"{OPENSEARCH_URL}/{INDEX_NAME}") except Exception: try: client.put(f"{OPENSEARCH_URL}/{INDEX_NAME}", json={ "settings": {"number_of_shards": 1, "number_of_replicas": 0}, "mappings": {"properties": { "url": {"type": "keyword"}, "title": {"type": "text"}, "content": {"type": "text"}, "excerpt": {"type": "text"}, "category": {"type": "keyword"}, "source_domain": {"type": "keyword"}, "crawled_at": {"type": "date"}, "indexed_at": {"type": "date"}, "metadata": {"type": "object"}, }} }) except Exception: pass def _ensure_qdrant(): try: client.get(f"{QDRANT_URL}/collections/{INDEX_NAME}") except Exception: try: client.put(f"{QDRANT_URL}/collections/{INDEX_NAME}", json={ "vectors": {"size": 768, "distance": "Cosine"} }) except Exception: pass def _ai_chat(prompt: str, system: str = "You are a research assistant. Be concise and factual.") -> str: r = ollama.post("/api/chat", json={ "model": OLLAMA_MODEL, "messages": [ {"role": "system", "content": system}, {"role": "user", "content": prompt}, ], "stream": False, "options": {"temperature": 0.3, "num_predict": 2048}, }) body = r.json() return body.get("message", {}).get("content", "") or body.get("thinking", "") or "" def _get_embedding(text: str) -> list: try: r = ollama.post("/api/embeddings", json={"model": "nomic-embed-text-v2-moe:latest", "prompt": text[:2048]}) return r.json().get("embedding", []) except Exception: return [] # ── Status ──────────────────────────────────────────────────── @app.get("/health") def health(): return {"status": "ok"} @app.get("/api/status") def status(): svc = {} try: r = client.get(f"{OPENSEARCH_URL}/_cluster/health") cnt = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_count").json() if r.status_code == 200 else {"count": 0} svc["opensearch"] = {"status": r.json().get("status", "down"), "documents": cnt.get("count", 0)} except Exception: svc["opensearch"] = {"status": "down"} try: r = client.get(f"{QDRANT_URL}/healthz") col = client.get(f"{QDRANT_URL}/collections/{INDEX_NAME}").json() svc["qdrant"] = {"status": "ok", "points": col.get("result", {}).get("points_count", 0)} except Exception: svc["qdrant"] = {"status": "down"} try: r = client.get(f"{YACY_URL}/yacysearch.json", params={"query": "test", "maximumRecords": 1}) svc["yacy"] = {"status": "ok" if r.status_code == 200 else "down"} except Exception: svc["yacy"] = {"status": "down"} try: r = ollama.get("/api/tags") svc["ollama"] = {"status": "ok", "models": len(r.json().get("models", []))} except Exception: svc["ollama"] = {"status": "down"} return {"services": svc} # ── Search ──────────────────────────────────────────────────── @app.get("/api/search") def search_web(q: str = Query(...), category: str = "", limit: int = 10): _ensure_index() body = { "size": limit, "query": {"bool": {"must": [{"multi_match": {"query": q, "fields": ["title^3", "content", "excerpt"]}}]}}, "highlight": {"fields": {"content": {"fragment_size": 200, "number_of_fragments": 2}}}, } if category: body["query"]["bool"]["filter"] = [{"term": {"category": category}}] try: r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json=body) result = r.json() hits = [] for h in result.get("hits", {}).get("hits", []): src = h["_source"] hits.append({"url": src.get("url"), "title": src.get("title"), "excerpt": src.get("excerpt") or (h.get("highlight", {}).get("content", [""])[0]), "category": src.get("category"), "crawled_at": src.get("crawled_at"), "score": h["_score"]}) return {"query": q, "total": result.get("hits", {}).get("total", {}).get("value", 0), "hits": hits} except Exception as e: return {"query": q, "total": 0, "hits": [], "note": f"Index may be empty. {e}"} @app.get("/api/semantic-search") def semantic_search(q: str = Query(...), limit: int = 10): _ensure_qdrant() emb = _get_embedding(q) if not emb: return {"hits": [], "error": "Embedding model not available"} try: r = client.post(f"{QDRANT_URL}/collections/{INDEX_NAME}/points/search", json={ "vector": emb, "limit": limit, "with_payload": True, "with_vector": False}) hits = [{"url": p.get("payload", {}).get("url"), "title": p.get("payload", {}).get("title"), "excerpt": str(p.get("payload", {}).get("excerpt", ""))[:300], "score": p.get("score")} for p in r.json().get("result", [])] return {"query": q, "total": len(hits), "hits": hits} except Exception as e: return {"hits": [], "error": str(e)} # ── Crawl ───────────────────────────────────────────────────── import re import html as html_mod def _fetch_and_index(url: str, category: str = ""): """Fetch a URL, extract text, and index into OpenSearch immediately.""" try: r = client.get(url, headers={"User-Agent": "Mozilla/5.0 (compatible; ResearchBot/1.0)"}, timeout=15.0) if r.status_code != 200: return None html_text = r.text # Basic HTML-to-text text = re.sub(r']*>.*?', '', html_text, flags=re.DOTALL|re.IGNORECASE) text = re.sub(r']*>.*?', '', text, flags=re.DOTALL|re.IGNORECASE) text = re.sub(r'<[^>]+>', ' ', text) text = re.sub(r'\s+', ' ', text).strip() text = html_mod.unescape(text) # Extract title title_match = re.search(r']*>(.*?)', html_text, re.IGNORECASE|re.DOTALL) title = html_mod.unescape(title_match.group(1).strip()) if title_match else url # Extract meta description desc_match = re.search(r']+name=["\']description["\'][^>]+content=["\']([^"\']+)', html_text, re.IGNORECASE) excerpt = desc_match.group(1)[:500] if desc_match else text[:500] # Derive domain + category from urllib.parse import urlparse domain = urlparse(url).netloc if not category: cat_map = {"wikipedia": "reference", "github": "software", "arxiv": "science", "docs.": "documentation", "blog.": "blog", "news.": "news"} for k, v in cat_map.items(): if k in domain: category = v break if not category: category = "web" # Index into OpenSearch import datetime doc = { "url": url, "title": title, "content": text[:50000], "excerpt": excerpt[:1000], "category": category, "source_domain": domain, "crawled_at": datetime.datetime.utcnow().isoformat(), "indexed_at": datetime.datetime.utcnow().isoformat(), } _ensure_index() client.put(f"{OPENSEARCH_URL}/{INDEX_NAME}/_doc/{hashlib.md5(url.encode()).hexdigest()}", json=doc, params={"refresh": "true"}) # Also index into Qdrant emb = _get_embedding(excerpt[:1000]) if emb: try: _ensure_qdrant() client.put(f"{QDRANT_URL}/collections/{INDEX_NAME}/points", json={ "points": [{"id": hashlib.md5(url.encode()).hexdigest(), "vector": emb, "payload": {"url": url, "title": title, "excerpt": excerpt[:500]}}] }) except Exception: pass return {"title": title, "domain": domain, "category": category, "size": len(text)} except Exception as e: return None @app.get("/api/crawl") def crawl_url(url: str = Query(...), depth: int = 1): # 1. Fetch and index immediately into OpenSearch + Qdrant indexed = _fetch_and_index(url) # 2. Also submit to YaCy for deeper crawling try: r = client.get(f"{YACY_URL}/Crawler_p.json", params={ "crawlingDomMaxPages": 50, "crawlingDepth": depth, "crawlingStart": url, "crawlingQ": "on", "bookmarkTitle": "research", "bookmarkFolder": "/research", "indexText": "on", "indexMedia": "on", "crawlingMode": "url", "cachePolicy": "iffresh", }) return {"status": "crawl_started", "url": url, "depth": depth, "indexed": indexed} except Exception as e: return {"status": "indexed_only", "url": url, "indexed": indexed, "yacy_error": str(e)} @app.get("/api/crawl-topic") def crawl_topic(topic: str = Query(...), max_urls: int = 20): discovered = [] try: r = client.get(f"{YACY_URL}/yacysearch.json", params={"query": topic, "maximumRecords": max_urls, "resource": "global"}) for ch in r.json().get("channels", []): for item in ch.get("items", []): if item.get("link"): discovered.append({"url": item["link"], "title": item.get("title", ""), "snippet": item.get("description", "")}) except Exception as e: return {"topic": topic, "error": f"Discovery failed: {e}", "urls_discovered": 0, "urls_crawled": 0} crawled = 0 for u in discovered[:max_urls]: try: client.get(f"{YACY_URL}/Crawler_p.json", params={ "crawlingDomMaxPages": 10, "crawlingDepth": 0, "crawlingStart": u["url"], "crawlingQ": "on", "indexText": "on", "indexMedia": "on", "crawlingMode": "url", "cachePolicy": "iffresh", }, timeout=5.0) crawled += 1 except Exception: pass return {"topic": topic, "urls_discovered": len(discovered), "urls_crawled": crawled} # ── Document Retrieval ──────────────────────────────────────── @app.get("/api/document") def retrieve_document(url: str = Query(...)): try: r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={ "size": 1, "query": {"term": {"url": url}}}) hits = r.json().get("hits", {}).get("hits", []) if not hits: return {"error": "Not found", "url": url} src = hits[0]["_source"] return {"url": src.get("url"), "title": src.get("title"), "content": src.get("content", "")[:10000], "excerpt": src.get("excerpt"), "category": src.get("category"), "crawled_at": src.get("crawled_at")} except Exception as e: return {"error": str(e), "url": url} # ── AI Synthesis ────────────────────────────────────────────── @app.get("/api/summarize") def summarize_sources(urls: str = Query(...), instruction: str = "Summarize key points"): url_list = [u.strip() for u in urls.split(",") if u.strip()] combined = "" for url in url_list[:5]: try: r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={"size": 1, "query": {"term": {"url": url}}}) hits = r.json().get("hits", {}).get("hits", []) if hits: src = hits[0]["_source"] combined += f"\n\n--- {url} ---\n{src.get('title','')}\n{src.get('content', src.get('excerpt',''))[:2000]}" except Exception: pass if not combined.strip(): return {"error": "No content retrieved"} summary = _ai_chat(f"Instruction: {instruction}\n\nSources:{combined}\n\nProvide a structured summary.") return {"instruction": instruction, "sources": len(url_list), "summary": summary} @app.get("/api/extract") def extract_information(url: str = Query(...), schema: str = Query("company names, products, prices")): try: r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={"size": 1, "query": {"term": {"url": url}}}) hits = r.json().get("hits", {}).get("hits", []) if not hits: return {"error": "Not found", "url": url} content = hits[0]["_source"].get("content", "")[:8000] prompt = f"Extract: {schema}\n\nDocument:\n{content}\n\nReturn ONLY valid JSON." result = _ai_chat(prompt, system="Extract structured data. Return ONLY valid JSON. No explanation.") return {"url": url, "schema": schema, "extracted": result} except Exception as e: return {"error": str(e)} @app.get("/api/report") def create_report(topic: str = Query(...), sources: str = ""): if sources: urls = [u.strip() for u in sources.split(",") if u.strip()] else: try: r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={ "size": 8, "query": {"multi_match": {"query": topic, "fields": ["title^3", "content"]}}}) urls = [h["_source"]["url"] for h in r.json().get("hits", {}).get("hits", [])] except Exception: urls = [] gathered = "" for url in urls[:8]: try: r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={"size": 1, "query": {"term": {"url": url}}}) hits = r.json().get("hits", {}).get("hits", []) if hits: src = hits[0]["_source"] gathered += f"\n\n### {src.get('title','Source')}\nURL: {url}\n{src.get('content',src.get('excerpt',''))[:1500]}" except Exception: pass prompt = f"""Research topic: {topic}\nSources:{gathered if gathered else ' No sources found.'} Generate a comprehensive report: 1. Executive Summary 2. Key Findings (numbered) 3. Source Analysis 4. Knowledge Gaps 5. Recommendations Be thorough, use markdown, cite sources.""" report = _ai_chat(prompt, system="You are a senior research analyst. Produce thorough, structured reports.") return {"topic": topic, "sources_used": len(urls), "report": report} # ── Research Pipeline ───────────────────────────────────────── @app.get("/api/research") def research_topic(topic: str = Query(...)): steps = [] kw_result = {"hits": []} sem_result = {"hits": []} try: r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={ "size": 5, "query": {"multi_match": {"query": topic, "fields": ["title^3", "content", "excerpt"]}}, "highlight": {"fields": {"content": {"fragment_size": 200, "number_of_fragments": 1}}}, }) hits = [{"title": h["_source"].get("title"), "excerpt": h.get("highlight", {}).get("content", [h["_source"].get("excerpt", "")])[0]} for h in r.json().get("hits", {}).get("hits", [])] kw_result = {"total": len(hits), "hits": hits} steps.append("keyword_search_done") except Exception: steps.append("keyword_search_skipped") try: emb = _get_embedding(topic) if emb: r = client.post(f"{QDRANT_URL}/collections/{INDEX_NAME}/points/search", json={ "vector": emb, "limit": 5, "with_payload": True}) hits2 = [{"title": p.get("payload", {}).get("title", ""), "excerpt": str(p.get("payload", {}).get("excerpt", ""))[:200], "score": p.get("score")} for p in r.json().get("result", [])] sem_result = {"total": len(hits2), "hits": hits2} steps.append("semantic_search_done") except Exception: steps.append("semantic_search_skipped") try: r = client.get(f"{YACY_URL}/yacysearch.json", params={"query": topic, "maximumRecords": 10, "resource": "global"}) discovered = [item.get("link") for ch in r.json().get("channels", []) for item in ch.get("items", []) if item.get("link")] for url in discovered[:10]: try: client.get(f"{YACY_URL}/Crawler_p.json", params={ "crawlingDomMaxPages": 10, "crawlingDepth": 0, "crawlingStart": url, "crawlingQ": "on", "indexText": "on", "indexMedia": "on", "crawlingMode": "url", "cachePolicy": "iffresh"}, timeout=5.0) except Exception: pass steps.append(f"crawl_dispatched_{len(discovered[:10])}") except Exception: steps.append("crawl_skipped") all_sources = "" for h in kw_result.get("hits", [])[:3] + sem_result.get("hits", [])[:3]: all_sources += f"- {h.get('title', 'Unknown')}: {h.get('excerpt', '')[:200]}\n" summary = "" if all_sources: summary = _ai_chat( f"Research topic: {topic}\n\nSources:\n{all_sources}\n\nConcise research summary (3-5 paragraphs): " "key findings, important sources, knowledge gaps, next steps.") return {"topic": topic, "steps": steps, "keyword_results": kw_result, "semantic_results": sem_result, "ai_summary": summary} # ── Dashboard ───────────────────────────────────────────────── @app.get("/", response_class=HTMLResponse) def dashboard(): return HTMLResponse(""" AI Research Engine

🧠 AI Research Engine

Self-hosted · Private

Your Private Research Cloud

Discover, index, and synthesize knowledge — all on your own infrastructure.

""") if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)