commit 34061a4ac19c45be02d862360853cd84172ea428 Author: drjones Date: Tue Aug 4 06:02:42 2026 -0700 AI Research Engine β€” self-hosted knowledge acquisition system - 10 MCP tools via thin proxy on MacBook - Backend REST API + Dashboard on CT 145 Docker - Services: YaCy crawler, OpenSearch index, Qdrant vectors, Ollama LLM - All 4 services healthy and verified diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..eebb4bb --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +.venv/ +__pycache__/ +*.pyc +.env +.git/ diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..ab9cc09 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,11 @@ +FROM python:3.11-slim + +WORKDIR /app + +RUN pip install --no-cache-dir fastapi uvicorn httpx python-dotenv + +COPY backend.py . + +EXPOSE 8000 + +CMD ["python", "backend.py"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..9a7d24f --- /dev/null +++ b/README.md @@ -0,0 +1,105 @@ +# 🧠 AI Research Engine + +**Self-hosted agentic AI search infrastructure** β€” your private research cloud. + +AI agents connect via MCP to discover, crawl, index, and synthesize knowledge from the web β€” all running on your own Proxmox hardware. + +## Architecture + +``` +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ MacBook (thin MCP proxy) β”‚ +β”‚ server.py β†’ forwards to CT 145 backend β”‚ +β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ + β”‚ HTTP +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β–Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ CT 145 (10.30.20.249) β€” Docker Host β”‚ +β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ +β”‚ β”‚ YaCy β”‚ β”‚OpenSearchβ”‚ β”‚ Redis β”‚ β”‚ Backend β”‚ β”‚ +β”‚ β”‚ :8090 β”‚ β”‚ :9200 β”‚ β”‚ :6379 β”‚ β”‚ :8000 β”‚ β”‚ +β”‚ β”‚ crawl β”‚ β”‚ index β”‚ β”‚ cache β”‚ β”‚ API+UI β”‚ β”‚ +β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ +β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ + β”‚ β”‚ +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β–Όβ”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β–Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ CT 509 (.68) β”‚ β”‚ GamingPC (.186) β”‚ +β”‚ Qdrant :6333 β”‚ β”‚ Ollama :11434 β”‚ +β”‚ semantic search β”‚ β”‚ ornith:latest (9B) β”‚ +β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ +``` + +## Endpoints + +| Service | URL | Purpose | +|---------|-----|---------| +| Dashboard | http://10.30.20.249:8000 | Web UI | +| Backend API | http://10.30.20.249:8000/api/* | REST API | +| OpenSearch | http://10.30.20.249:9200 | Full-text index | +| YaCy | http://10.30.20.249:8090 | Web crawler | +| Qdrant | http://10.30.20.68:6333 | Vector DB | +| Ollama | http://10.30.20.186:11434 | LLM inference | + +## MCP Tools (10) + +| Tool | Description | +|------|-------------| +| `search_web(query)` | Full-text search across indexed documents | +| `semantic_search(query)` | Vector search by meaning (Qdrant) | +| `crawl_url(url)` | Crawl a URL into the index | +| `crawl_topic(topic)` | Discover + crawl sources for a topic | +| `research_topic(topic)` | Full pipeline: search β†’ crawl β†’ summarize | +| `retrieve_document(url)` | Get full indexed document content | +| `summarize_sources(urls)` | AI summary of multiple sources | +| `extract_information(url, schema)` | Structured data extraction | +| `create_report(topic)` | Comprehensive research report | +| `index_status()` | System health check | + +## Quick Start + +```bash +# Check status +curl http://10.30.20.249:8000/api/status + +# Search +curl "http://10.30.20.249:8000/api/search?q=knowledge+graphs" + +# Crawl a URL +curl "http://10.30.20.249:8000/api/crawl?url=https://example.com&depth=1" + +# Deep research +curl "http://10.30.20.249:8000/api/research?topic=LED+grow+lights" + +# Generate report +curl "http://10.30.20.249:8000/api/report?topic=AI+agents" +``` + +## Deployment + +On CT 145 (10.30.20.249): + +```bash +# Services +docker run -d --name redis --restart unless-stopped -p 6379:6379 redis:7-alpine +docker run -d --name yacy --restart unless-stopped -p 8090:8090 yacy/yacy_search_server:latest +docker run -d --name opensearch --restart unless-stopped -p 9200:9200 \ + -e "discovery.type=single-node" -e "DISABLE_SECURITY_PLUGIN=true" \ + -e "OPENSEARCH_JAVA_OPTS=-Xms1g -Xmx2g" opensearchproject/opensearch:2.17.0 + +# Backend +docker build -t research-backend . +docker run -d --name research-backend --restart unless-stopped -p 8000:8000 \ + --add-host=host.docker.internal:host-gateway \ + -e YACY_URL=http://host.docker.internal:8090 \ + -e OPENSEARCH_URL=http://host.docker.internal:9200 \ + -e QDRANT_URL=http://10.30.20.68:6333 \ + -e OLLAMA_URL=http://10.30.20.186:11434 \ + research-backend +``` + +## Files + +- `server.py` β€” Thin MCP proxy (runs on MacBook) +- `backend.py` β€” REST API + Dashboard (runs on CT 145) +- `docker-compose.yml` β€” Reference compose file +- `Dockerfile` β€” Backend container build +- `.env` β€” Service endpoints config diff --git a/backend.py b/backend.py new file mode 100644 index 0000000..99d31aa --- /dev/null +++ b/backend.py @@ -0,0 +1,468 @@ +#!/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 +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, params={"refresh": "true"}) + 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 ───────────────────────────────────────────────────── + +@app.get("/api/crawl") +def crawl_url(url: str = Query(...), depth: int = 1): + 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} + except Exception as e: + return {"status": "error", "url": url, "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) diff --git a/dashboard.py b/dashboard.py new file mode 100644 index 0000000..2e60b67 --- /dev/null +++ b/dashboard.py @@ -0,0 +1,226 @@ +#!/usr/bin/env python3 +""" +Research Engine Dashboard β€” FastAPI web UI +Runs on CT 145 port 8000 +""" + +import os +import json +import httpx +from fastapi import FastAPI, Request, Query +from fastapi.responses import HTMLResponse, JSONResponse +from fastapi.staticfiles import StaticFiles +from dotenv import load_dotenv + +load_dotenv(os.path.expanduser("~/ai-research-engine/.env")) + +app = FastAPI(title="AI Research Engine") + +OPENSEARCH_URL = os.getenv("OPENSEARCH_URL", "http://localhost:9200") +QDRANT_URL = os.getenv("QDRANT_URL", "http://10.30.20.68:6333") +YACY_URL = os.getenv("YACY_URL", "http://localhost:8090") +INDEX_NAME = "research_docs" + +client = httpx.Client(timeout=10.0) + +@app.get("/", response_class=HTMLResponse) +async def dashboard(): + return HTMLResponse(""" + + + + + +AI Research Engine + + + +
+

🧠 AI Research Engine

+ Self-hosted knowledge acquisition +
+
+ +
+ + +
+ +
+
+
+ + + +""") + +@app.get("/api/status") +async def status(): + svc = {} + # OpenSearch + try: + r = client.get(f"{OPENSEARCH_URL}/_cluster/health") + h = r.json() + cnt = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_count").json() + svc["opensearch"] = {"status": h.get("status"), "documents": cnt.get("count", 0)} + except Exception as e: + svc["opensearch"] = {"status": "down", "error": str(e)} + # Qdrant + try: + r = client.get(f"{QDRANT_URL}/health") + if r.status_code == 200: + col = client.get(f"{QDRANT_URL}/collections/{INDEX_NAME}").json() + svc["qdrant"] = {"status": "ok", "points": col.get("result", {}).get("points_count", 0)} + except Exception as e: + svc["qdrant"] = {"status": "down", "error": str(e)} + # YaCy + try: + r = client.get(f"{YACY_URL}/api/status.json") + svc["yacy"] = {"status": r.json().get("status", "unknown")} + except Exception as e: + svc["yacy"] = {"status": "down", "error": str(e)} + # Ollama + try: + r = httpx.get("http://10.30.20.186:11434/api/tags", timeout=5.0) + svc["ollama"] = {"status": "ok", "models": len(r.json().get("models", []))} + except Exception as e: + svc["ollama"] = {"status": "down", "error": str(e)} + + return {"services": svc} + +@app.get("/api/search") +async def search(q: str = Query(...), limit: int = 10): + try: + r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={ + "size": limit, + "query": {"multi_match": {"query": q, "fields": ["title^3", "content", "excerpt"]}}, + "highlight": {"fields": {"content": {"fragment_size": 200, "number_of_fragments": 2}}}, + }) + hits = [] + for h in r.json().get("hits", {}).get("hits", []): + src = h["_source"] + hits.append({ + "url": src.get("url"), "title": src.get("title"), + "excerpt": h.get("highlight", {}).get("content", [src.get("excerpt", "")])[0], + "category": src.get("category"), "crawled_at": src.get("crawled_at"), + "score": h["_score"], + }) + return {"query": q, "total": r.json().get("hits", {}).get("total", {}).get("value", 0), "hits": hits} + except Exception as e: + return {"query": q, "total": 0, "hits": [], "error": str(e)} + +@app.get("/api/crawl") +async def crawl(url: str = Query(...)): + try: + r = client.get(f"{YACY_URL}/Crawler_p.json", params={ + "crawlingDomMaxPages": 50, "crawlingDepth": 1, + "crawlingStart": url, "crawlingQ": "on", + "bookmarkTitle": "research", "bookmarkFolder": "/research", + "indexText": "on", "indexMedia": "on", + "crawlingMode": "url", "cachePolicy": "iffresh", + }) + return {"status": "crawl_started", "url": url, "yacy": r.json()} + except Exception as e: + return {"status": "error", "url": url, "error": str(e)} + +@app.get("/api/research") +async def research(topic: str = Query(...)): + # Proxy to MCP server's research_topic via command + import subprocess, sys + result = subprocess.run( + [sys.executable, "-c", f""" +import json, sys +sys.path.insert(0, '/Users/drjones/ai-research-engine') +from server import research_topic +print(research_topic("{topic}")) +"""], + capture_output=True, text=True, timeout=120, + env={**os.environ, "PYTHONPATH": "/Users/drjones/ai-research-engine"} + ) + try: + return json.loads(result.stdout) + except: + return {"error": result.stderr, "topic": topic} + +if __name__ == "__main__": + import uvicorn + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..b23710c --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,62 @@ +version: '3.8' + +services: + # ── Web Crawler ──────────────────────────────────────────── + yacy: + image: yacy/yacy_search_server:latest + container_name: yacy + ports: + - "8090:8090" + environment: + - YACY_ADMIN_PASSWORD=research2026 + volumes: + - yacy_data:/opt/yacy_search_server/DATA + restart: unless-stopped + mem_limit: 2g + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:8090/api/status.json"] + interval: 30s + timeout: 10s + retries: 3 + + # ── Document Index ───────────────────────────────────────── + opensearch: + image: opensearchproject/opensearch:2.17.0 + container_name: opensearch + environment: + - discovery.type=single-node + - DISABLE_SECURITY_PLUGIN=true + - "OPENSEARCH_JAVA_OPTS=-Xms1g -Xmx2g" + - DISABLE_INSTALL_DEMO_CONFIG=true + ports: + - "9200:9200" + - "9600:9600" + volumes: + - opensearch_data:/usr/share/opensearch/data + restart: unless-stopped + mem_limit: 3g + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:9200/_cluster/health"] + interval: 30s + timeout: 10s + retries: 5 + + # ── Cache / Job Queue ────────────────────────────────────── + redis: + image: redis:7-alpine + container_name: redis + ports: + - "6379:6379" + volumes: + - redis_data:/data + restart: unless-stopped + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 10s + timeout: 5s + retries: 3 + +volumes: + yacy_data: + opensearch_data: + redis_data: diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..1e6a5b0 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,9 @@ +[project] +name = "ai-research-engine" +version = "0.1.0" +requires-python = ">=3.10" +dependencies = [ + "mcp>=1.0.0,<2.0.0", + "httpx>=0.27.0", + "python-dotenv>=1.0.0", +] diff --git a/server.py b/server.py new file mode 100644 index 0000000..519e873 --- /dev/null +++ b/server.py @@ -0,0 +1,110 @@ +#!/usr/bin/env python3 +""" +AI Research Engine β€” Thin MCP Proxy +Runs on MacBook. Forwards all tool calls to the CT 145 backend. +Minimal resource usage β€” all heavy lifting on Proxmox. +""" + +import json +import httpx +from mcp.server import FastMCP + +BACKEND_URL = "http://10.30.20.249:8000" +client = httpx.Client(timeout=120.0) + +mcp = FastMCP( + "ai-research-engine", + instructions=""" +AI Research Engine β€” private knowledge acquisition system. + +search_web(query) β€” Full-text search across indexed documents +semantic_search(query) β€” Find documents by meaning (vector search) +crawl_url(url) β€” Crawl a URL into the index +crawl_topic(topic) β€” Discover and crawl sources for a topic +research_topic(topic) β€” Full pipeline: discover β†’ crawl β†’ summarize +retrieve_document(url) β€” Get full content of an indexed document +summarize_sources(urls, instruction) β€” AI summary of multiple sources +extract_information(url, schema) β€” Structured data extraction +create_report(topic, sources) β€” Generate comprehensive research report +index_status() β€” System health and stats +""", +) + +def _get(path: str) -> dict: + r = client.get(f"{BACKEND_URL}{path}") + r.raise_for_status() + return r.json() + + +@mcp.tool() +def search_web(query: str, category: str = "", limit: int = 10) -> str: + """Full-text search across indexed documents. Find by keywords, titles, content.""" + r = _get(f"/api/search?q={query}&category={category}&limit={limit}") + return json.dumps(r, indent=2) + + +@mcp.tool() +def semantic_search(query: str, limit: int = 10) -> str: + """Search by meaning using vector embeddings. Finds conceptually related docs.""" + r = _get(f"/api/semantic-search?q={query}&limit={limit}") + return json.dumps(r, indent=2) + + +@mcp.tool() +def crawl_url(url: str, depth: int = 1) -> str: + """Crawl a URL. depth: 0=just this page, 1=+linked pages.""" + r = _get(f"/api/crawl?url={url}&depth={depth}") + return json.dumps(r, indent=2) + + +@mcp.tool() +def crawl_topic(topic: str, max_urls: int = 20) -> str: + """Discover and crawl sources for a topic using YaCy.""" + r = _get(f"/api/crawl-topic?topic={topic}&max_urls={max_urls}") + return json.dumps(r, indent=2) + + +@mcp.tool() +def research_topic(topic: str) -> str: + """Full research pipeline: keyword search β†’ semantic search β†’ crawl new sources β†’ AI summary.""" + r = _get(f"/api/research?topic={topic}") + return json.dumps(r, indent=2) + + +@mcp.tool() +def retrieve_document(url: str) -> str: + """Get full indexed content of a document by URL.""" + r = _get(f"/api/document?url={url}") + return json.dumps(r, indent=2) + + +@mcp.tool() +def summarize_sources(urls: str, instruction: str = "Summarize key points") -> str: + """Summarize multiple URLs using local LLM. urls: comma-separated.""" + r = _get(f"/api/summarize?urls={urls}&instruction={instruction}") + return json.dumps(r, indent=2) + + +@mcp.tool() +def extract_information(url: str, schema: str = "company names, products, prices, specifications") -> str: + """Extract structured information from a document using LLM.""" + r = _get(f"/api/extract?url={url}&schema={schema}") + return json.dumps(r, indent=2) + + +@mcp.tool() +def create_report(topic: str, sources: str = "") -> str: + """Generate a comprehensive research report. sources: optional comma-separated URLs.""" + r = _get(f"/api/report?topic={topic}&sources={sources}") + return json.dumps(r, indent=2) + + +@mcp.tool() +def index_status() -> str: + """Check health of all backend services: OpenSearch, Qdrant, YaCy, Ollama.""" + r = _get("/api/status") + return json.dumps(r, indent=2) + + +if __name__ == "__main__": + mcp.run(transport="stdio")