Deploy GraphRAG benchmark backend

.gitattributes

@@ -33,3 +33,5 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+*.sqlite3 filter=lfs diff=lfs merge=lfs -text
+*.json filter=lfs diff=lfs merge=lfs -text

backend/Dockerfile

@@ -0,0 +1,27 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# system deps (small set; keep slim)
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt /app/requirements.txt
+RUN pip install --no-cache-dir -r /app/requirements.txt
+
+# Copy code (keep repo structure)
+COPY backend /app/backend
+COPY ingestion /app/ingestion
+COPY evaluation /app/evaluation
+COPY pipelines /app/pipelines
+COPY utils /app/utils
+COPY services /app/services
+COPY scripts /app/scripts
+
+ENV PYTHONUNBUFFERED=1
+
+EXPOSE 8000
+
+# Render sets $PORT; local default remains 8000
+CMD ["sh", "-c", "uvicorn backend.main:app --host 0.0.0.0 --port ${PORT:-8000}"]

backend/main.py

@@ -0,0 +1,82 @@
+import os
+from pathlib import Path
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi import FastAPI
+
+from backend.routes import ingestion, metrics, query
+from backend.routes.graph import router as graph_router
+from backend.security import (
+    path_writable,
+    production_config_errors,
+    rate_limit_middleware,
+    request_size_middleware,
+)
+from utils.paths import chroma_path, graph_path, upload_dir
+
+app = FastAPI(title="GraphRAG Benchmark API")
+
+frontend_url = os.getenv("FRONTEND_URL", "").strip()
+cors_origins_env = os.getenv("CORS_ORIGINS", frontend_url).strip()
+cors_origins = [o.strip() for o in cors_origins_env.split(",") if o.strip()]
+allow_origin_regex = os.getenv("CORS_ORIGIN_REGEX")  # optional
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=cors_origins if cors_origins else ["http://localhost:3000", "http://localhost:3005"],
+    allow_origin_regex=allow_origin_regex,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+app.middleware("http")(rate_limit_middleware)
+app.middleware("http")(request_size_middleware)
+
+app.include_router(ingestion.router, prefix="/api")
+app.include_router(metrics.router, prefix="/api")
+app.include_router(query.router, prefix="/api")
+app.include_router(graph_router)
+
+
+@app.on_event("startup")
+def validate_startup_config():
+    errors = production_config_errors()
+    if errors:
+        raise RuntimeError("; ".join(errors))
+
+
+@app.get("/health")
+def health_check():
+    return {"status": "ok"}
+
+
+@app.get("/ready")
+def ready_check():
+    graph_file = Path(graph_path())
+    checks = {
+        "production_config": production_config_errors(),
+        "chroma_path_writable": path_writable(chroma_path()),
+        "upload_dir_writable": path_writable(upload_dir()),
+        "graph_artifact_present": graph_file.exists(),
+        "tigergraph_enabled": os.getenv("TIGERGRAPH_ENABLED", "false").lower() == "true",
+        "networkx_primary": True,
+    }
+    if checks["tigergraph_enabled"]:
+        checks["tigergraph_configured"] = all(
+            os.getenv(key)
+            for key in (
+                "TIGERGRAPH_HOST",
+                "TIGERGRAPH_GRAPH",
+                "TIGERGRAPH_USERNAME",
+                "TIGERGRAPH_PASSWORD",
+            )
+        )
+    else:
+        checks["tigergraph_configured"] = "skipped"
+
+    ready = (
+        not checks["production_config"]
+        and checks["chroma_path_writable"]
+        and checks["upload_dir_writable"]
+        and (checks["tigergraph_configured"] is True or checks["tigergraph_configured"] == "skipped")
+    )
+    return {"status": "ready" if ready else "not_ready", "checks": checks}

backend/routes/graph.py

@@ -0,0 +1,63 @@
+from fastapi import APIRouter
+import pickle
+import os
+
+from utils.paths import graph_path
+
+router = APIRouter(prefix="/api/graph", tags=["graph"])
+
+
+@router.get("/view")
+def get_graph_view(limit: int = 200):
+    graph_file = graph_path()
+    if not os.path.exists(graph_file):
+        return {
+            "status": "error",
+            "message": "Graph file not found",
+            "nodes": [],
+            "edges": []
+        }
+
+    with open(graph_file, "rb") as f:
+        graph = pickle.load(f)
+
+    nodes = []
+    edges = []
+
+    for node_id, attrs in list(graph.nodes(data=True))[:limit]:
+        nodes.append({
+            "data": {
+                "id": str(node_id),
+                "label": attrs.get("label")
+                    or attrs.get("name")
+                    or attrs.get("title")
+                    or attrs.get("text", "")[:40]
+                    or str(node_id),
+                "type": attrs.get("node_type", attrs.get("type", "unknown"))
+            }
+        })
+
+    node_ids = set(n["data"]["id"] for n in nodes)
+
+    for source, target, attrs in graph.edges(data=True):
+        if str(source) in node_ids and str(target) in node_ids:
+            edges.append({
+                "data": {
+                    "id": f"{source}-{target}",
+                    "source": str(source),
+                    "target": str(target),
+                    "label": attrs.get("relation", attrs.get("type", "RELATED"))
+                }
+            })
+
+    return {
+        "status": "success",
+        "nodes": nodes,
+        "edges": edges,
+        "stats": {
+            "total_nodes": graph.number_of_nodes(),
+            "total_edges": graph.number_of_edges(),
+            "shown_nodes": len(nodes),
+            "shown_edges": len(edges)
+        }
+    }

backend/routes/ingestion.py

@@ -0,0 +1,34 @@
+from fastapi import APIRouter, UploadFile, File, HTTPException
+
+from backend.security import AdminDependency
+
+router = APIRouter()
+
+
+@router.post("/upload")
+async def upload_file(file: UploadFile = File(...), _: None = AdminDependency):
+    allowed = (".pdf", ".txt", ".csv")
+    if not any(file.filename.lower().endswith(ext) for ext in allowed):
+        raise HTTPException(status_code=400, detail="Only PDF, TXT, CSV supported")
+
+    content = await file.read()
+
+    from backend.services.ingestion_service import ingest_document
+
+    try:
+        result = await ingest_document(file.filename, content)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    except Exception as exc:
+        raise HTTPException(
+            status_code=500,
+            detail={
+                "stage": "ingestion",
+                "message": "Ingestion failed",
+            },
+        ) from exc
+
+    return {
+        "status": result.get("status", "success"),
+        "data": result,
+    }

backend/routes/metrics.py

@@ -0,0 +1,22 @@
+import json
+from pathlib import Path
+
+from fastapi import APIRouter
+
+router = APIRouter()
+
+FINAL_SUMMARY_PATH = Path("data/results/final_summary.json")
+
+
+@router.get("/metrics/final-summary")
+def final_summary():
+    if not FINAL_SUMMARY_PATH.exists():
+        return {"status": "missing", "summary": {}}
+
+    try:
+        return {
+            "status": "ok",
+            "summary": json.loads(FINAL_SUMMARY_PATH.read_text(encoding="utf-8")),
+        }
+    except json.JSONDecodeError:
+        return {"status": "error", "summary": {}, "error": "Invalid final_summary.json"}

backend/routes/query.py

@@ -0,0 +1,19 @@
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel, Field
+
+from backend.services.pipelines_service import run_all_pipelines
+
+router = APIRouter()
+
+
+class QueryRequest(BaseModel):
+    query: str = Field(..., min_length=1)
+
+
+@router.post("/query")
+def query_all(request: QueryRequest):
+    question = request.query.strip()
+    if not question:
+        raise HTTPException(status_code=400, detail="Query cannot be empty")
+
+    return run_all_pipelines(question)

backend/routes/tigergraph.py

@@ -0,0 +1,38 @@
+from fastapi import APIRouter
+from pydantic import BaseModel
+
+from pipelines.graphrag.graphrag_client import TigerGraphClient
+
+router = APIRouter()
+
+
+class TigerGraphHealth(BaseModel):
+    ok: bool
+    host: str
+    graph: str
+    restpp_version_ok: bool = False
+    graph_exists: bool = False
+    error: str | None = None
+
+
+@router.get("/tigergraph/health", response_model=TigerGraphHealth)
+def tigergraph_health():
+    client = TigerGraphClient()
+
+    health = TigerGraphHealth(ok=False, host=client.host, graph=client.graph)
+
+    try:
+        # Use RESTPP /version which does not require a graph.
+        version = client._get_version()
+        health.restpp_version_ok = bool(version) and not version.get("error", False)
+
+        # Check the benchmark graph exists by listing one vertex type with limit=1.
+        # If graph does not exist, RESTPP returns an error.
+        client._get("/vertices/BenchDocument", params={"limit": 1})
+        health.graph_exists = True
+
+        health.ok = health.restpp_version_ok and health.graph_exists
+        return health
+    except Exception as exc:
+        health.error = str(exc)
+        return health

backend/security.py

@@ -0,0 +1,92 @@
+import os
+import time
+from collections import defaultdict, deque
+from pathlib import Path
+from typing import Callable
+
+from fastapi import Depends, HTTPException, Request, status
+from fastapi.responses import JSONResponse
+
+
+MAX_REQUEST_BYTES = int(os.getenv("MAX_REQUEST_BYTES", str(25 * 1024 * 1024)))
+RATE_LIMIT_REQUESTS = int(os.getenv("RATE_LIMIT_REQUESTS", "120"))
+RATE_LIMIT_WINDOW_SECONDS = int(os.getenv("RATE_LIMIT_WINDOW_SECONDS", "60"))
+
+_requests: dict[str, deque[float]] = defaultdict(deque)
+
+
+def admin_auth_required() -> bool:
+    return bool(os.getenv("ADMIN_API_KEY")) or os.getenv("ENV", "").lower() == "production"
+
+
+def require_admin(request: Request) -> None:
+    expected = os.getenv("ADMIN_API_KEY")
+    if not admin_auth_required():
+        return
+    if not expected:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="ADMIN_API_KEY is required in production.",
+        )
+
+    auth = request.headers.get("authorization", "")
+    if not auth.lower().startswith("bearer "):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token")
+    token = auth.split(" ", 1)[1].strip()
+    if token != expected:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Invalid bearer token")
+
+
+AdminDependency = Depends(require_admin)
+
+
+async def request_size_middleware(request: Request, call_next: Callable):
+    content_length = request.headers.get("content-length")
+    if content_length and int(content_length) > MAX_REQUEST_BYTES:
+        return JSONResponse(
+            status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
+            content={"detail": "Request body too large"},
+        )
+    return await call_next(request)
+
+
+async def rate_limit_middleware(request: Request, call_next: Callable):
+    client = request.client.host if request.client else "unknown"
+    now = time.time()
+    window_start = now - RATE_LIMIT_WINDOW_SECONDS
+    bucket = _requests[client]
+    while bucket and bucket[0] < window_start:
+        bucket.popleft()
+    if len(bucket) >= RATE_LIMIT_REQUESTS:
+        return JSONResponse(
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            content={"detail": "Rate limit exceeded"},
+        )
+    bucket.append(now)
+    return await call_next(request)
+
+
+def production_config_errors() -> list[str]:
+    if os.getenv("ENV", "").lower() != "production":
+        return []
+
+    errors = []
+    for key in ("OPENAI_API_KEY", "FRONTEND_URL", "BACKEND_URL", "ADMIN_API_KEY"):
+        if not os.getenv(key):
+            errors.append(f"{key} is required in production")
+    if not (os.getenv("JWT_SECRET") or os.getenv("SESSION_SECRET")):
+        errors.append("JWT_SECRET or SESSION_SECRET is required in production")
+    return errors
+
+
+def path_writable(path: str) -> bool:
+    p = Path(path)
+    target = p if p.suffix == "" else p.parent
+    try:
+        target.mkdir(parents=True, exist_ok=True)
+        probe = target / ".write_probe"
+        probe.write_text("ok", encoding="utf-8")
+        probe.unlink(missing_ok=True)
+        return True
+    except Exception:
+        return False

backend/services/ingestion_service.py

@@ -0,0 +1,88 @@
+import uuid
+import os
+from pathlib import Path
+
+from ingestion.preprocess import extract_text_from_file
+from ingestion.chunk_data import chunk_text
+from ingestion.build_embeddings import add_to_vector_store
+from ingestion.entity_extraction import extract_entities
+from ingestion.build_graph import build_graph
+from utils.paths import upload_dir
+
+
+async def ingest_document(file_name: str, file_bytes: bytes):
+    """
+    Unified ingestion for:
+    - Basic RAG (vector store)
+    - GraphRAG (local NetworkX graph)
+    """
+
+    doc_id = str(uuid.uuid4())
+
+    # Persist raw uploads (useful for debugging / re-processing). This does not affect pipeline outputs.
+    try:
+        up_dir = upload_dir()
+        os.makedirs(up_dir, exist_ok=True)
+        safe_name = Path(file_name).name.replace("\\", "_").replace("/", "_")
+        raw_path = os.path.join(up_dir, f"{doc_id}__{safe_name}")
+        with open(raw_path, "wb") as f:
+            f.write(file_bytes)
+    except Exception:
+        # Non-fatal: ingestion should still proceed.
+        pass
+
+    # -------------------------
+    # 1. PREPROCESS
+    # -------------------------
+    pages, file_type = extract_text_from_file(file_name, file_bytes)
+    if not pages or not any(p.strip() for p in pages):
+        raise ValueError("No extractable text found in file")
+
+    # -------------------------
+    # 2. CHUNKING
+    # -------------------------
+    chunk_records = []
+    chunk_index = 0
+    for page_num, page_text in enumerate(pages, start=1):
+        if not page_text.strip():
+            continue
+        page_chunks = chunk_text(page_text)
+        for chunk in page_chunks:
+            chunk_records.append(
+                {
+                    "chunk_id": f"{doc_id}_chunk_{chunk_index}",
+                    "doc_id": doc_id,
+                    "text": chunk,
+                    "source_file": file_name,
+                    "page": page_num if file_type == "pdf" else None,
+                }
+            )
+            chunk_index += 1
+
+    # -------------------------
+    # 3. EMBEDDING + STORE (ONLY ONCE)
+    # -------------------------
+    add_to_vector_store(chunk_records)
+
+    # -------------------------
+    # 4. GRAPH RAG PIPELINE
+    # -------------------------
+    entities = extract_entities([c["text"] for c in chunk_records])
+
+    graph_result = build_graph(
+        doc_id=doc_id,
+        title=file_name,
+        chunks=chunk_records,
+        entities=entities
+    )
+
+    # -------------------------
+    # RETURN METADATA
+    # -------------------------
+    return {
+        "status": "success",
+        "doc_id": doc_id,
+        "chunks_count": len(chunk_records),
+        "entities_count": len(entities),
+        "relations_count": graph_result.get("relations_edges", 0),
+    }

backend/services/pipelines_service.py

@@ -0,0 +1,112 @@
+import json
+import os
+from importlib import import_module
+from pathlib import Path
+from typing import Callable, Dict
+
+
+GROUND_TRUTH_PATH = Path("evaluation/ground_truth.json")
+
+
+def _load_reference_answer(question: str) -> str:
+    if not GROUND_TRUTH_PATH.exists():
+        return ""
+
+    try:
+        data = json.loads(GROUND_TRUTH_PATH.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return ""
+
+    if isinstance(data, dict):
+        return data.get(question, "")
+
+    if isinstance(data, list):
+        for row in data:
+            if not isinstance(row, dict):
+                continue
+            row_question = row.get("question", row.get("query", ""))
+            if row_question == question:
+                return row.get("correct_answer", row.get("answer", ""))
+
+    return ""
+
+
+def _safe_run(name: str, runner: Callable[[str], Dict], question: str) -> Dict:
+    try:
+        result = runner(question)
+        if not isinstance(result, dict):
+            return {
+                "status": "error",
+                "answer": "",
+                "error": f"{name} returned {type(result).__name__}, expected dict",
+            }
+
+        return {
+            "status": "success",
+            **result,
+        }
+    except Exception as exc:
+        return {
+            "status": "error",
+            "answer": "",
+            "error": str(exc),
+        }
+
+
+def _run_pipeline(name: str, module_path: str, function_name: str, question: str) -> Dict:
+    try:
+        module = import_module(module_path)
+        runner = getattr(module, function_name)
+    except Exception as exc:
+        return {
+            "status": "error",
+            "answer": "",
+            "error": f"Failed to load {name}: {exc}",
+        }
+
+    return _safe_run(name, runner, question)
+
+
+def run_all_pipelines(question: str) -> Dict:
+    """
+    Lazily import pipelines so the API can boot even when optional runtime
+    dependencies, model files, or external services are not ready yet.
+    """
+    response = {
+        "query": question,
+        "pipelines": {
+            "llm_only": _run_pipeline(
+                "llm_only",
+                "pipelines.llm_only.llm_pipeline",
+                "run_llm_only",
+                question,
+            ),
+            "basic_rag": _run_pipeline(
+                "basic_rag",
+                "pipelines.basic_rag.rag_pipeline",
+                "run_basic_rag",
+                question,
+            ),
+            "graphrag": _run_pipeline(
+                "graphrag",
+                "pipelines.graphrag.graphrag_pipeline",
+                "run_graphrag",
+                question,
+            ),
+        },
+    }
+
+    if os.getenv("ENABLE_LIVE_ACCURACY", "").lower() in {"1", "true", "yes"}:
+        correct_answer = _load_reference_answer(question)
+        if correct_answer:
+            from evaluation.evaluator import evaluate_single_answer
+
+            for result in response["pipelines"].values():
+                if result.get("status") == "success":
+                    result["accuracy"] = evaluate_single_answer(
+                        question,
+                        correct_answer,
+                        result.get("answer", ""),
+                    )
+
+    return response

data/chroma/12e587cb-7739-45a4-bb36-b1fba0ef511c/data_level0.bin

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:436726978ef423c6bf7020dbaefd11604b8f4b725fbe7ae7459c2b45f6ccd4db
+size 6764336

data/chroma/12e587cb-7739-45a4-bb36-b1fba0ef511c/header.bin

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:07205de30b67ddebefe7a25e1874717fd92d7d75e0fec53f4bcb0edf2fcea994
+size 100

data/chroma/12e587cb-7739-45a4-bb36-b1fba0ef511c/index_metadata.pickle

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:fdf9384ba9da795eedc06a313c97b519652b320e9fff1c9abe2fd765315e6a96
+size 258448

data/chroma/12e587cb-7739-45a4-bb36-b1fba0ef511c/length.bin

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:b9c4aa95b8a4c614039830f12760d7e4793fbd5f638feb215f2ad06bdf7ea2b3
+size 16144

data/chroma/12e587cb-7739-45a4-bb36-b1fba0ef511c/link_lists.bin

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:3b82f70ae47004403898f0f8fdd8e40f36a687bb917a3f7e68bf258c8dd27007
+size 34368

data/chroma/chroma.sqlite3

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:022f2b2e044df94587b5b6de6ebb0874d0857fa205c62fe4e50cfac9d5e1f105
+size 72962048

data/eval/scientific_eval_questions.json

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:1f4fdad1a16718a12085c81771a9b76550c852ee2d740b708712a0a9231883cb
+size 31464

data/graph/graph_metadata.json

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:16bd6977675d8a9096ba4968066526b015855159c6b4e44014d6c58bcf111cb8
+size 113

data/graph/graphrag_graph.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:a3b675daf240d1a0bfe7fe1bfad4f05eabf6c8a718c0254bf87ceed461e19683
+size 37461540

data/results/final_summary.json

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:76b04c2f30efdb9924e0ab0c01b508251cd2bd76a8f0858ac8183dcbcfc36e98
+size 1140

data/results/scientific_accuracy_report.json

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:a5cca828fecdf0b28ca03a5cc4fe059c9f3975e176ea9bf894d42fe34759f0b5
+size 377

data/results/scientific_benchmark_results.json

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:e27bff660544f05e8f9006dc6716c2398ed98faeb77f6c4d4543a3d2da1ae055
+size 176379

evaluation/bertscore_eval.py

@@ -0,0 +1,49 @@
+def compute_bertscore(predictions, references, lang="en"):
+    pairs = [
+        (prediction or "", reference or "")
+        for prediction, reference in zip(predictions, references)
+        if reference
+    ]
+    if not pairs:
+        return {
+            "f1": [],
+            "mean_f1": None,
+            "status": "SKIP",
+            "error": "No reference answers supplied.",
+        }
+
+    try:
+        import evaluate
+    except ImportError:
+        return {
+            "f1": [],
+            "mean_f1": None,
+            "status": "SKIP",
+            "error": "Install the evaluate package to compute BERTScore.",
+        }
+
+    filtered_predictions = [prediction for prediction, _ in pairs]
+    filtered_references = [reference for _, reference in pairs]
+    try:
+        bertscore = evaluate.load("bertscore")
+        result = bertscore.compute(
+            predictions=filtered_predictions,
+            references=filtered_references,
+            lang=lang,
+            rescale_with_baseline=True,
+        )
+    except Exception as exc:
+        return {
+            "f1": [],
+            "mean_f1": None,
+            "status": "SKIP",
+            "error": str(exc),
+        }
+
+    f1 = result.get("f1", [])
+    return {
+        "f1": f1,
+        "mean_f1": sum(f1) / len(f1) if f1 else None,
+        "status": "OK",
+        "error": None,
+    }

evaluation/evaluator.py

@@ -0,0 +1,61 @@
+from .bertscore_eval import compute_bertscore
+from .llm_judge import judge_answer, judge_answers
+from .metrics import compute_cost
+
+def evaluate(query, outputs, ground_truth):
+    results = {}
+
+    for name, out in outputs.items():
+        answer = out.get("answer", "")
+        tokens = out.get("tokens", 0)
+        latency = out.get("latency", 0)
+        results[name] = {
+            "answer": answer,
+            "tokens": tokens,
+            "latency": latency,
+            "cost": compute_cost(tokens),
+            "judge": judge_answer(answer, ground_truth, query)
+        }
+
+    return results
+
+
+def evaluate_single_answer(question, correct_answer, system_answer):
+    verdict = judge_answer(system_answer, correct_answer, question)
+    bert = compute_bertscore([system_answer], [correct_answer])
+    return {
+        "llm_judge": verdict,
+        "llm_judge_pass": verdict == "PASS",
+        "bertscore_f1": bert["mean_f1"],
+    }
+
+
+def evaluate_batch(pipeline_answers, ground_truth):
+    references = [row.get("correct_answer", "") for row in ground_truth]
+    questions = [row.get("question", row.get("query", "")) for row in ground_truth]
+    metrics = {}
+
+    for pipeline_name, answers in pipeline_answers.items():
+        rows = [
+            {
+                "question": question,
+                "correct_answer": reference,
+                "system_answer": answer,
+            }
+            for question, reference, answer in zip(questions, references, answers)
+        ]
+        verdicts = judge_answers(rows)
+        pass_fail = [verdict == "PASS" for verdict in verdicts if verdict != "SKIP"]
+        bert = compute_bertscore(answers, references)
+
+        metrics[pipeline_name] = {
+            "llm_judge_pass_rate": (
+                sum(pass_fail) / len(pass_fail) if pass_fail else None
+            ),
+            "llm_judge_verdicts": verdicts,
+            "bertscore_f1": bert["mean_f1"],
+            "bertscore_status": bert["status"],
+            "bertscore_error": bert["error"],
+        }
+
+    return metrics

evaluation/ground_truth.json

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:37517e5f3dc66819f61f5a7bb8ace1921282415f10551d2defa5c3eb0985b570
+size 3

evaluation/llm_judge.py

@@ -0,0 +1,76 @@
+import os
+
+JUDGE_MODEL = os.getenv("HF_JUDGE_MODEL", "meta-llama/Llama-3.1-8B-Instruct")
+
+JUDGE_PROMPT = """Grade the system's answer.
+Question: {question}
+Correct answer: {correct_answer}
+System answer: {system_answer}
+
+Reply with only PASS or FAIL.
+PASS = the system answer correctly addresses the question with no major errors.
+FAIL = the answer is wrong, missing, or contradicts the correct answer."""
+
+
+def _verdict_from_text(text):
+    normalized = (text or "").strip().upper()
+    if normalized.startswith("PASS") or "PASS" in normalized:
+        return "PASS"
+    return "FAIL"
+
+
+def judge_answer(system_answer, correct_answer, question="", client=None):
+    if not correct_answer:
+        return "SKIP"
+
+    if client is None:
+        token = os.getenv("HF_TOKEN")
+        if not token:
+            return "SKIP"
+
+        try:
+            from huggingface_hub import InferenceClient
+        except ImportError:
+            return "SKIP"
+
+        client = InferenceClient(model=JUDGE_MODEL, token=token)
+
+    prompt = JUDGE_PROMPT.format(
+        question=question,
+        correct_answer=correct_answer,
+        system_answer=system_answer or "",
+    )
+    try:
+        response = client.chat_completion(
+            [{"role": "user", "content": prompt}],
+            max_tokens=10,
+            temperature=0.0,
+        )
+    except Exception:
+        return "SKIP"
+
+    return _verdict_from_text(response.choices[0].message.content)
+
+
+def judge_answers(rows, model=JUDGE_MODEL):
+    token = os.getenv("HF_TOKEN")
+    if not token:
+        return ["SKIP" for _ in rows]
+
+    try:
+        from huggingface_hub import InferenceClient
+    except ImportError:
+        return ["SKIP" for _ in rows]
+
+    client = InferenceClient(model=model, token=token)
+    verdicts = []
+    for row in rows:
+        verdicts.append(
+            judge_answer(
+                row.get("system_answer", ""),
+                row.get("correct_answer", ""),
+                row.get("question", ""),
+                client=client,
+            )
+        )
+    return verdicts

evaluation/metrics.py

@@ -0,0 +1,2 @@
+def compute_cost(tokens, cost_per_1k=0.002):
+    return (tokens/1000) * cost_per_1k

ingestion/build_embeddings.py

@@ -0,0 +1,13 @@
+from pipelines.basic_rag.vector_store import VectorStore
+
+
+def add_to_vector_store(chunk_records):
+    """
+    Real-time ingestion into persisted ChromaDB
+    """
+    if not chunk_records:
+        raise ValueError("Cannot build embeddings for an empty document")
+
+    store = VectorStore.load()
+    store.add_documents(chunk_records)
+    return None

ingestion/build_graph.py

@@ -0,0 +1,81 @@
+from collections import defaultdict
+
+from pipelines.graphrag.graphrag_client import NetworkXGraphClient, normalize_text
+from ingestion.entity_extraction import extract_entities
+
+MAX_ENTITIES_PER_CHUNK = 12
+
+
+def build_graph(doc_id: str, title: str, chunks: list[dict], entities: list[tuple[str, str]]):
+    """
+    Local GraphRAG graph builder using NetworkXGraphClient.
+
+    Nodes:
+    - Document
+    - Chunk
+    - Entity
+
+    Edges:
+    - Document -> Chunk: HAS_CHUNK
+    - Chunk -> Entity: MENTIONS
+    - Entity -> Entity: RELATED_TO (co-mentions, capped per chunk)
+    """
+    client = NetworkXGraphClient()
+    client.load_graph()
+
+    source_file = chunks[0].get("source_file", title) if chunks else title
+    client.add_document(doc_id, title=title, source_file=source_file)
+
+    entity_type_map = {normalize_text(name): entity_type for name, entity_type in entities}
+
+    relations_edges = 0
+    mentions_edges = 0
+    co_mentions: dict[tuple[str, str], set[str]] = defaultdict(set)
+
+    for chunk in chunks:
+        chunk_id = chunk["chunk_id"]
+        text = chunk["text"]
+        client.add_chunk(chunk_id=chunk_id, doc_id=doc_id, text=text)
+
+        # Extract entities from this chunk, cap to avoid relation explosion.
+        chunk_entities_raw = [name for name, _ in extract_entities([text])]
+        chunk_entities: list[str] = []
+        seen = set()
+        for name in chunk_entities_raw:
+            ent = normalize_text(name)
+            if not ent or ent in seen:
+                continue
+            chunk_entities.append(ent)
+            seen.add(ent)
+            if len(chunk_entities) >= MAX_ENTITIES_PER_CHUNK:
+                break
+
+        # Add entity nodes + mentions edges.
+        for ent in chunk_entities:
+            client.add_entity(ent, entity_type=entity_type_map.get(ent, "Concept"))
+            client.add_mentions_edge(chunk_id=chunk_id, entity_id=ent)
+            mentions_edges += 1
+
+        # Record co-mentions for RELATED_TO edges.
+        unique = sorted(set(chunk_entities))
+        for i in range(len(unique)):
+            for j in range(i + 1, len(unique)):
+                a, b = unique[i], unique[j]
+                co_mentions[(a, b)].add(chunk_id)
+
+    for (a, b), evidence_chunks in co_mentions.items():
+        # Store one evidence chunk for this relation (keeps edges lean).
+        evidence = next(iter(evidence_chunks)) if evidence_chunks else None
+        client.add_related_edge(a, b, evidence_chunk_id=evidence)
+        relations_edges += 1
+
+    client.save_graph()
+
+    return {
+        "documents": 1,
+        "chunks": len(chunks),
+        "entities": len(entities),
+        "mentions_edges": mentions_edges,
+        "relations_edges": relations_edges,
+    }
+

ingestion/build_graph_tigergraph.py

@@ -0,0 +1,38 @@
+from pipelines.graphrag.graphrag_client import get_connection
+from ingestion.entity_extraction import extract_entities
+
+
+def build_graph(doc_id, title, chunks, entities):
+    """
+    Legacy TigerGraph graph builder (kept for future compatibility).
+    Not used in the local NetworkX GraphRAG pivot.
+    """
+    conn = get_connection()
+    mention_edges = 0
+    entity_ids = {entity_name for entity_name, _ in entities}
+
+    conn.upsert_document(doc_id, title)
+
+    for chunk in chunks:
+        conn.upsert_chunk(chunk["chunk_id"], chunk["text"])
+        conn.link_doc_chunk(doc_id, chunk["chunk_id"])
+
+    for entity_name, entity_type in entities:
+        entity_id = entity_name.lower()
+        conn.upsert_entity(entity_id, entity_name, entity_type)
+
+    for chunk in chunks:
+        chunk_entities = extract_entities([chunk["text"]])
+        for entity_name, _ in chunk_entities:
+            if entity_name in entity_ids:
+                conn.link_chunk_entity(chunk["chunk_id"], entity_name)
+                mention_edges += 1
+
+    return {
+        "documents": 1,
+        "chunks": len(chunks),
+        "entities": len(entities),
+        "has_chunk_edges": len(chunks),
+        "mentions_edges": mention_edges,
+    }
+

ingestion/chunk_data.py

@@ -0,0 +1,10 @@
+def chunk_text(text, chunk_size=500, overlap=50):
+    chunks = []
+
+    start = 0
+    while start < len(text):
+        end = start + chunk_size
+        chunks.append(text[start:end])
+        start += chunk_size - overlap
+
+    return chunks

ingestion/entity_extraction.py

@@ -0,0 +1,47 @@
+import re
+
+STOPWORDS = {
+    "about",
+    "after",
+    "also",
+    "because",
+    "being",
+    "between",
+    "could",
+    "does",
+    "from",
+    "for",
+    "have",
+    "into",
+    "more",
+    "other",
+    "paper",
+    "than",
+    "that",
+    "their",
+    "there",
+    "these",
+    "this",
+    "through",
+    "using",
+    "what",
+    "when",
+    "where",
+    "which",
+    "with",
+}
+
+
+def extract_entities(chunks):
+    entities = set()
+
+    for text in chunks:
+        words = re.findall(r"\b[A-Z][A-Za-z0-9]*(?:-[A-Z0-9][A-Za-z0-9]*)*\b", text)
+        words.extend(re.findall(r"\b[a-z][a-z0-9-]{4,}\b", text))
+
+        for w in words:
+            normalized = w.strip("-").lower()
+            if normalized and normalized not in STOPWORDS:
+                entities.add((normalized, "Concept"))
+
+    return list(entities)

ingestion/preprocess.py

@@ -0,0 +1,38 @@
+import fitz  # PyMuPDF
+
+def extract_text_from_pdf(file_bytes):
+    doc = fitz.open(stream=file_bytes, filetype="pdf")
+
+    text = ""
+    for page in doc:
+        text += page.get_text()
+
+    return text
+
+
+def extract_pages_from_pdf(file_bytes):
+    doc = fitz.open(stream=file_bytes, filetype="pdf")
+    pages = []
+    for page in doc:
+        pages.append(page.get_text())
+    return pages
+
+
+def extract_text_from_text_bytes(file_bytes: bytes) -> str:
+    for enc in ("utf-8", "utf-8-sig", "latin-1"):
+        try:
+            return file_bytes.decode(enc)
+        except Exception:
+            continue
+    # fallback: replace undecodable bytes
+    return file_bytes.decode("utf-8", errors="replace")
+
+
+def extract_text_from_file(file_name: str, file_bytes: bytes):
+    name = (file_name or "").lower()
+    if name.endswith(".pdf"):
+        return extract_pages_from_pdf(file_bytes), "pdf"
+    if name.endswith(".txt") or name.endswith(".csv"):
+        return [extract_text_from_text_bytes(file_bytes)], "text"
+    # default: try as text
+    return [extract_text_from_text_bytes(file_bytes)], "text"

ingestion/schema.py

@@ -0,0 +1,62 @@
+import os
+
+from dotenv import load_dotenv
+from pyTigerGraph import TigerGraphConnection
+
+load_dotenv()
+
+
+def create_schema():
+    conn = TigerGraphConnection(
+        host=os.getenv("TG_GSQL_HOST", "http://localhost"),
+        graphname="",
+        username=os.getenv("TG_USERNAME", "tigergraph"),
+        password=os.getenv("TG_PASSWORD", "tigergraph"),
+        restppPort=os.getenv("TG_RESTPP_PORT", "9000"),
+        gsPort=os.getenv("TG_GSQL_PORT", "14240"),
+    )
+
+    graph = os.getenv("TG_GRAPH", "graphrag_benchmark")
+
+    commands = [
+        """
+CREATE VERTEX BenchDocument (
+  PRIMARY_ID doc_id STRING,
+  title STRING
+) WITH primary_id_as_attribute="true"
+""",
+        """
+CREATE VERTEX BenchChunk (
+  PRIMARY_ID chunk_id STRING,
+  text STRING
+) WITH primary_id_as_attribute="true"
+""",
+        """
+CREATE VERTEX BenchEntity (
+  PRIMARY_ID entity_id STRING,
+  name STRING,
+  entity_type STRING
+) WITH primary_id_as_attribute="true"
+""",
+        "CREATE UNDIRECTED EDGE BenchHasChunk (FROM BenchDocument, TO BenchChunk)",
+        "CREATE UNDIRECTED EDGE BenchMentions (FROM BenchChunk, TO BenchEntity)",
+        "CREATE UNDIRECTED EDGE BenchRelated (FROM BenchEntity, TO BenchEntity)",
+        f"CREATE GRAPH {graph} (BenchDocument, BenchChunk, BenchEntity, BenchHasChunk, BenchMentions, BenchRelated)",
+    ]
+
+    output = []
+    for command in commands:
+        try:
+            result = conn.gsql(command)
+        except Exception as exc:
+            message = str(exc)
+            if "already exists" not in message and "existed" not in message:
+                raise
+            result = message
+        output.append(result)
+
+    return "\n".join(output)
+
+
+if __name__ == "__main__":
+    print(create_schema())

pipelines/basic_rag/chunking.py

@@ -0,0 +1,13 @@
+import re
+
+def chunk_text(text, chunk_size=2):
+    # Split into sentences
+    sentences = re.split(r'(?<=[.!?]) +', text)
+
+    chunks = []
+
+    for i in range(0, len(sentences), chunk_size):
+        chunk = " ".join(sentences[i:i + chunk_size])
+        chunks.append(chunk.strip())
+
+    return chunks

pipelines/basic_rag/embedding.py

@@ -0,0 +1,14 @@
+MODEL_NAME = "all-MiniLM-L6-v2"
+model = None
+
+
+def get_model():
+    global model
+    if model is None:
+        from sentence_transformers import SentenceTransformer
+
+        model = SentenceTransformer(MODEL_NAME)
+    return model
+
+def embed_text(texts):
+    return get_model().encode(texts)

pipelines/basic_rag/rag_pipeline.py

@@ -0,0 +1,102 @@
+from .retriever import retrieve
+from .vector_store import VectorStore
+from utils.llm import generate
+
+VECTOR_STORE = None
+MAX_CONTEXT_CHARS = 2500
+OVERVIEW_TERMS = (
+    "what is this paper about",
+    "what is the paper about",
+    "summarize the paper",
+    "summary of the paper",
+    "paper summary",
+    "abstract",
+    "main idea",
+)
+
+
+def get_store():
+    global VECTOR_STORE
+    if VECTOR_STORE is None:
+        VECTOR_STORE = VectorStore.load()
+    return VECTOR_STORE
+
+
+def _chunk_number(record):
+    chunk_id = record.get("chunk_id", "")
+    try:
+        return int(chunk_id.rsplit("_chunk_", 1)[1])
+    except (IndexError, ValueError):
+        return 10**9
+
+
+def _is_overview_query(query):
+    normalized = " ".join(query.lower().split())
+    return any(term in normalized for term in OVERVIEW_TERMS)
+
+
+def _overview_chunks(store, limit=3):
+    # Chroma doesn't expose raw metadata list like the old FAISS store.
+    # Overview queries will rely on semantic retrieval for now.
+    return []
+
+
+def _merge_chunks(primary, secondary, limit=5):
+    merged = []
+    seen = set()
+
+    for record in [*primary, *secondary]:
+        text = record.get("text", "")
+        dedupe_key = " ".join(text.split()).lower() or record.get("chunk_id")
+        if dedupe_key not in seen:
+            merged.append(record)
+            seen.add(dedupe_key)
+
+        if len(merged) == limit:
+            break
+
+    return merged
+
+
+def run_basic_rag(query):
+    store = get_store()
+    semantic_chunks = retrieve(query, store, k=3)
+    if _is_overview_query(query):
+        retrieved_chunks = _merge_chunks(_overview_chunks(store), semantic_chunks)
+    else:
+        retrieved_chunks = semantic_chunks
+
+    context = "\n\n".join(
+        f"[{i + 1}] {chunk.get('text', '')}"
+        for i, chunk in enumerate(retrieved_chunks)
+    )
+    context = context[:MAX_CONTEXT_CHARS]
+
+    prompt = f"""
+You are an AI assistant.
+
+Use ONLY the provided context to answer the question.
+If the context does not contain enough information, say so.
+
+Context:
+{context}
+
+Question:
+{query}
+
+Answer:
+"""
+
+    res = generate(prompt)
+
+    return {
+        "answer": res["text"],
+        "context": context,
+        "tokens": res["total_tokens"],
+        "latency": res["latency"],
+        "details": {
+            "prompt_tokens": res["prompt_tokens"],
+            "completion_tokens": res["completion_tokens"],
+            "retrieved_chunks": retrieved_chunks,
+        },
+    }

pipelines/basic_rag/retriever.py

@@ -0,0 +1,19 @@
+def retrieve(query, vector_store, k=3):
+    results = vector_store.search(query, k=k * 10)
+
+    unique = []
+    seen = set()
+
+    for record in results:
+        chunk_id = record.get("chunk_id")
+        text = record.get("text", "")
+        dedupe_key = " ".join(text.split()).lower() or chunk_id
+
+        if dedupe_key not in seen:
+            unique.append(record)
+            seen.add(dedupe_key)
+
+        if len(unique) == k:
+            break
+
+    return unique

pipelines/basic_rag/vector_store.py

@@ -0,0 +1,93 @@
+import os
+from typing import Any, Dict, List, Optional
+
+import chromadb
+
+from pipelines.basic_rag.embedding import embed_text
+from utils.paths import chroma_path
+
+COLLECTION_NAME = "chunks"
+
+
+class VectorStore:
+    """
+    ChromaDB-backed persistent vector store.
+
+    Public API (used by ingestion/pipelines):
+    - add_documents(chunk_records)
+    - search(query, k=5, filters=None)
+
+    Notes:
+    - Uses chunk_id as the Chroma record id.
+    - Stores chunk text as document, plus metadata fields.
+    """
+
+    def __init__(self):
+        chroma_dir = chroma_path()
+        os.makedirs(chroma_dir, exist_ok=True)
+        self.client = chromadb.PersistentClient(path=chroma_dir)
+        self.collection = self.client.get_or_create_collection(name=COLLECTION_NAME)
+
+    @classmethod
+    def load(cls, path: Optional[str] = None):
+        # Keep signature stable; `path` is unused for Chroma.
+        return cls()
+
+    def add_documents(self, chunk_records: List[Dict[str, Any]]) -> int:
+        if not chunk_records:
+            return 0
+
+        ids: List[str] = []
+        documents: List[str] = []
+        metadatas: List[Dict[str, Any]] = []
+
+        for record in chunk_records:
+            chunk_id = record["chunk_id"]
+            ids.append(chunk_id)
+            documents.append(record["text"])
+            metadatas.append(
+                {
+                    "doc_id": record.get("doc_id", ""),
+                    "chunk_id": chunk_id,
+                    "source_file": record.get("source_file", ""),
+                    "page": record.get("page"),
+                }
+            )
+
+        embeddings = embed_text(documents)
+        embeddings_list = [e.tolist() for e in embeddings]
+
+        # Chroma raises if ids already exist. For ingestion re-runs, upsert.
+        self.collection.upsert(
+            ids=ids,
+            documents=documents,
+            embeddings=embeddings_list,
+            metadatas=metadatas,
+        )
+        return len(ids)
+
+    def search(self, query: str, k: int = 5, filters: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
+        query_embedding = embed_text([query])[0].tolist()
+
+        result = self.collection.query(
+            query_embeddings=[query_embedding],
+            n_results=k,
+            where=filters,
+            include=["documents", "metadatas", "distances"],
+        )
+
+        documents = (result.get("documents") or [[]])[0]
+        metadatas = (result.get("metadatas") or [[]])[0]
+        distances = (result.get("distances") or [[]])[0]
+
+        out: List[Dict[str, Any]] = []
+        for doc, meta, dist in zip(documents, metadatas, distances):
+            record = dict(meta or {})
+            record["text"] = doc
+            record["score"] = float(dist) if dist is not None else None
+            out.append(record)
+        return out
+
+    # Back-compat helpers (old FAISS codepaths)
+    def search_text(self, query: str, k: int = 5):
+        return self.search(query, k=k)