Add Hugging Face README config properly encoded

Dockerfile

@@ -0,0 +1,18 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# Install dependencies
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy the rest of the application
+COPY . .
+
+# Hugging Face Spaces and Koyeb use port 7860 or 8000. 
+# We'll use 7860 as it's the HF default.
+ENV PORT=7860
+EXPOSE 7860
+
+# Run the FastAPI server
+CMD uvicorn backend.main:app --host 0.0.0.0 --port $PORT

README.md

@@ -0,0 +1,10 @@
+---
+title: Knowledge Loom Backend
+emoji: 📚
+colorFrom: blue
+colorTo: green
+sdk: docker
+pinned: false
+---
+
+This is the FastAPI backend for Knowledge Loom.

backend/agents/aggregator_agent.py

@@ -0,0 +1,68 @@
+"""
+AGGREGATOR AGENT
+----------------
+Job: Take the raw outputs of the specialist agents and merge them into
+one clean, well-formatted Markdown study report.
+
+Tool used: Gemini for text formatting and synthesis only.
+"""
+
+
+def run_aggregator_agent(model, topic: str, overview: dict, videos: dict,
+                         papers: dict) -> str:
+    video_lines = "\n".join(
+        f"- [{v['title']}]({v['url']}) - {v['channel']}"
+        for v in videos.get("videos", [])
+    ) or "_No videos found._"
+
+    paper_lines = "\n".join(
+        f"- **{p['title']}** ({p['year']}) - {p['authors']} - "
+        f"{p['citations']} citations. [Link]({p['url']})"
+        for p in papers.get("papers", [])
+    ) or "_No papers found._"
+
+    prompt = f"""
+You are the final editor assembling a research study guide on the topic: "{topic}".
+
+Combine the material below into a single clean Markdown report with this
+exact structure. Do not add facts that are not present in the material.
+Lightly polish wording only; do not rewrite the overview's substance.
+
+# {topic}
+
+## Overview
+{overview.get('content', '')}
+
+## Recommended Videos
+{video_lines}
+
+## Research Paper Suggestions
+{paper_lines}
+
+---
+Return the final Markdown report only, nothing else.
+"""
+
+    try:
+        response = model["client"].models.generate_content(
+            model=model["model_name"], contents=prompt
+        )
+        return response.text.strip()
+    except Exception as e:
+        # Programmatic backup formatting if the Gemini service fails (e.g., due to a 429 rate limit)
+        fallback_report = f"""# {topic}
+
+## Overview
+{overview.get('content', '')}
+
+## Recommended Videos
+{video_lines}
+
+## Research Paper Suggestions
+{paper_lines}
+
+---
+*Note: This report was assembled using the system's rule-based programmatic backup aggregator because the Gemini LLM synthesis service is currently rate-limited ({e}).*
+"""
+        return fallback_report.strip()
+

backend/agents/overview_agent.py

@@ -0,0 +1,98 @@
+"""
+Agent 1: OVERVIEW AGENT
+------------------------
+Job: Given a topic, produce a clear, structured explanation:
+     - what it is
+     - why it matters
+     - 4-6 key sub-concepts a learner should know
+
+Tool used: Gemini with Google Search grounding (falls back to plain
+generation if grounding isn't available in the SDK version installed).
+
+Only holds the GEMINI_API_KEY — no other credential is passed in.
+"""
+
+from google import genai
+from google.genai import types
+
+
+def build_overview_agent(api_key: str, model_name: str):
+    """Returns a dict bundling the client + model name (the 'agent')."""
+    client = genai.Client(api_key=api_key)
+    return {"client": client, "model_name": model_name}
+
+
+def run_overview_agent(agent, topic: str) -> dict:
+    client = agent["client"]
+    model_name = agent["model_name"]
+
+    prompt = (
+        f"You are an educational research assistant. The user wants to learn "
+        f"about: \"{topic}\".\n\n"
+        "Produce a well-structured overview with these exact sections:\n"
+        "1. **What it is** (2-3 sentences, plain language)\n"
+        "2. **Why it matters** (2-3 sentences)\n"
+        "3. **Key concepts to understand** (4-6 bullet points, each one line)\n"
+        "4. **Common misconceptions** (1-2 bullet points, if any exist)\n\n"
+        "Be accurate and concise. Do not invent facts. If you are uncertain "
+        "about something, say so rather than guessing."
+    )
+
+    import time
+    
+    # Simple retry helper to tolerate rate limits (429)
+    response = None
+    grounded = False
+    last_error = None
+    
+    for attempt in range(3):
+        try:
+            # Attempt Google Search grounding so the overview is based on
+            # current web results, not just the model's training data.
+            response = client.models.generate_content(
+                model=model_name,
+                contents=prompt,
+                config=types.GenerateContentConfig(
+                    tools=[types.Tool(google_search=types.GoogleSearch())]
+                ),
+            )
+            grounded = True
+            break
+        except Exception as e:
+            last_error = e
+            if "429" in str(e) or "RESOURCE_EXHAUSTED" in str(e):
+                time.sleep(2 * (attempt + 1))  # Exponential backoff
+                continue
+            
+            # If not a 429, try a fallback: plain generation without grounding
+            try:
+                response = client.models.generate_content(
+                    model=model_name, contents=prompt
+                )
+                grounded = False
+                break
+            except Exception as inner_e:
+                last_error = inner_e
+                if "429" in str(inner_e) or "RESOURCE_EXHAUSTED" in str(inner_e):
+                    time.sleep(2 * (attempt + 1))
+                    continue
+                break
+
+    if response is not None:
+        return {
+            "agent": "overview_agent",
+            "grounded_with_search": grounded,
+            "content": response.text.strip(),
+        }
+    else:
+        # Graceful degradation if Gemini key is fully exhausted/blocked
+        return {
+            "agent": "overview_agent",
+            "grounded_with_search": False,
+            "content": (
+                f"_AI-generated overview is temporarily unavailable (Gemini API quota exhausted/rate limit hit). "
+                f"Please verify your API key limits. Topic requested: \"{topic}\"_"
+            ),
+            "error": str(last_error)
+        }
+

backend/agents/paper_agent.py

@@ -0,0 +1,103 @@
+"""
+Agent 3: PAPER AGENT
+---------------------
+Job: Given a topic, find the most important / most-cited research papers.
+
+Tool used: OpenAlex API. It is free, structured, and does not require an
+API key. Results are sorted by citation count.
+"""
+
+import re
+import time
+
+import requests
+
+OPENALEX_WORKS_URL = "https://api.openalex.org/works"
+OPENALEX_MAILTO = "noreply@edu-research-agent.local"
+
+
+def _abstract_from_inverted_index(inverted_index: dict | None) -> str:
+    if not inverted_index:
+        return ""
+
+    positions = []
+    for word, indexes in inverted_index.items():
+        for index in indexes:
+            positions.append((index, word))
+
+    return " ".join(word for _, word in sorted(positions))
+
+
+def _get_openalex(params: dict, headers: dict) -> dict:
+    params = {**params, "mailto": OPENALEX_MAILTO}
+    last_error = None
+
+    for wait_seconds in (0, 2, 5):
+        if wait_seconds:
+            time.sleep(wait_seconds)
+
+        try:
+            resp = requests.get(
+                OPENALEX_WORKS_URL, params=params, headers=headers, timeout=10
+            )
+            resp.raise_for_status()
+            return resp.json()
+        except requests.HTTPError as e:
+            last_error = e
+            if e.response is None or e.response.status_code != 429:
+                raise
+        except requests.RequestException as e:
+            last_error = e
+            raise
+
+    raise last_error
+
+
+def run_paper_agent(api_key: str, topic: str, max_results: int = 10) -> dict:
+    query = re.sub(r"\bner\b", "named entity recognition", topic, flags=re.IGNORECASE)
+    params = {
+        "filter": f"title_and_abstract.search:{query}",
+        "per-page": max_results,
+        "sort": "cited_by_count:desc",
+        "select": (
+            "display_name,authorships,publication_year,cited_by_count,"
+            "abstract_inverted_index,doi,id,primary_location"
+        ),
+    }
+    headers = {"User-Agent": "edu-research-agent/1.0"}
+
+    try:
+        data = _get_openalex(params, headers)
+    except requests.RequestException as e:
+        return {"agent": "paper_agent", "papers": [], "error": str(e)}
+
+    if not data.get("results"):
+        params.pop("filter", None)
+        params["search"] = query
+        try:
+            data = _get_openalex(params, headers)
+        except requests.RequestException as e:
+            return {"agent": "paper_agent", "papers": [], "error": str(e)}
+
+    cleaned = []
+    for paper in data.get("results", [])[:max_results]:
+        authors = ", ".join(
+            item.get("author", {}).get("display_name", "")
+            for item in (paper.get("authorships") or [])[:3]
+        )
+        location = paper.get("primary_location") or {}
+        source = location.get("source") or {}
+        url = location.get("landing_page_url") or paper.get("doi") or paper.get("id")
+        abstract = _abstract_from_inverted_index(paper.get("abstract_inverted_index"))
+
+        cleaned.append({
+            "title": paper.get("display_name"),
+            "authors": authors or "Unknown",
+            "year": paper.get("publication_year"),
+            "citations": paper.get("cited_by_count"),
+            "venue": source.get("display_name"),
+            "url": url,
+            "abstract": abstract[:300],
+        })
+
+    return {"agent": "paper_agent", "papers": cleaned, "error": None}

backend/agents/video_agent.py

@@ -0,0 +1,56 @@
+"""
+Agent 2: VIDEO AGENT
+---------------------
+Job: Given a topic, find the most relevant YouTube videos.
+
+Tool used: YouTube Data API v3 (search.list endpoint) — a structured,
+official API call. No scraping, no visiting youtube.com directly.
+
+Only holds the YOUTUBE_API_KEY.
+"""
+
+import requests
+
+YOUTUBE_SEARCH_URL = "https://www.googleapis.com/youtube/v3/search"
+
+
+def run_video_agent(api_key: str, topic: str, max_results: int = 5) -> dict:
+    if not api_key or "YOUR_YOUTUBE_API_KEY_HERE" in api_key:
+        return {
+            "agent": "video_agent",
+            "videos": [],
+            "error": "YouTube API key not configured.",
+        }
+
+    params = {
+        "part": "snippet",
+        "q": f"{topic} explained",
+        "type": "video",
+        "maxResults": max_results,
+        "order": "relevance",
+        "safeSearch": "strict",
+        "key": api_key,
+    }
+
+    try:
+        resp = requests.get(YOUTUBE_SEARCH_URL, params=params, timeout=10)
+        resp.raise_for_status()
+        data = resp.json()
+    except requests.RequestException as e:
+        return {"agent": "video_agent", "videos": [], "error": str(e)}
+
+    videos = []
+    for item in data.get("items", []):
+        video_id = item.get("id", {}).get("videoId")
+        snippet = item.get("snippet", {})
+        if not video_id:
+            continue
+        videos.append({
+            "title": snippet.get("title"),
+            "channel": snippet.get("channelTitle"),
+            "published": snippet.get("publishedAt"),
+            "url": f"https://www.youtube.com/watch?v={video_id}",
+            "thumbnail": snippet.get("thumbnails", {}).get("medium", {}).get("url"),
+        })
+
+    return {"agent": "video_agent", "videos": videos, "error": None}

backend/config.py

@@ -0,0 +1,30 @@
+"""
+config.py
+---------
+Central place where required API keys are loaded.
+"""
+
+import os
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+# Gemini API key - used by Overview Agent, Aggregator Agent, and Safety Gate.
+GEMINI_API_KEY = os.getenv("GEMINI_API_KEY", "YOUR_GEMINI_API_KEY_HERE")
+
+# YouTube Data API v3 key - used by Video Agent.
+YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY", "YOUR_YOUTUBE_API_KEY_HERE")
+
+# Model name used for Gemini calls.
+GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.0-flash")
+
+
+def validate_keys():
+    """Warn if required keys are still placeholders."""
+    missing = []
+    if "YOUR_GEMINI_API_KEY_HERE" in GEMINI_API_KEY:
+        missing.append("GEMINI_API_KEY")
+    if "YOUR_YOUTUBE_API_KEY_HERE" in YOUTUBE_API_KEY:
+        missing.append("YOUTUBE_API_KEY")
+    return missing

backend/main.py

@@ -0,0 +1,50 @@
+"""
+main.py
+-------
+FastAPI backend exposing a single endpoint:
+
+    POST /research
+    body: {"topic": "reinforcement learning"}
+    returns: {status, reason, report, agent_log}
+
+Run with:
+    uvicorn backend.main:app --reload --port 8000
+"""
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+
+from backend.orchestrator import run_research_pipeline
+from backend import config
+
+app = FastAPI(title="Education Research Agent API")
+
+# Allow the Streamlit frontend (running on a different port) to call this API
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+class ResearchRequest(BaseModel):
+    topic: str
+
+
+@app.get("/health")
+def health_check():
+    missing = config.validate_keys()
+    return {
+        "status": "ok",
+        "missing_required_keys": missing,
+    }
+
+
+@app.post("/research")
+def research(request: ResearchRequest):
+    topic = request.topic.strip()
+    if not topic:
+        return {"status": "error", "reason": "Empty topic.", "report": None, "agent_log": []}
+    return run_research_pipeline(topic)

backend/orchestrator.py

@@ -0,0 +1,102 @@
+"""
+orchestrator.py
+----------------
+This is the "multi-agent system" coordination layer.
+
+Flow:
+  1. Keyword pre-filter on the raw topic (fast, cheap safety gate)
+  2. Run Overview, Video, and Paper agents CONCURRENTLY (they don't
+     depend on each other's output)
+  3. Pass all 3 results into the Aggregator Agent to merge into one report
+  4. Run an LLM-based safety check on the FINAL report before returning it
+  5. Return structured status per agent + the final report
+"""
+
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
+from backend import config
+from backend.security import keyword_prefilter, llm_safety_check
+from backend.agents.overview_agent import build_overview_agent, run_overview_agent
+from backend.agents.video_agent import run_video_agent
+from backend.agents.paper_agent import run_paper_agent
+from backend.agents.aggregator_agent import run_aggregator_agent
+
+
+def run_research_pipeline(topic: str) -> dict:
+    # --- Step 1: fast pre-filter -------------------------------------------------
+    if not keyword_prefilter(topic):
+        return {
+            "status": "blocked",
+            "reason": "Topic blocked by pre-filter safety rule.",
+            "report": None,
+            "agent_log": [],
+        }
+
+    agent_log = []
+
+    # Shared Gemini model instance for Overview + Aggregator + Safety check.
+    # (Video/Paper agents use their own separate, scoped API keys.)
+    gemini_model = build_overview_agent(config.GEMINI_API_KEY, config.GEMINI_MODEL)
+
+    # --- Step 2: run the 3 specialist agents concurrently ------------------------
+    results = {}
+    with ThreadPoolExecutor(max_workers=3) as executor:
+        futures = {
+            executor.submit(run_overview_agent, gemini_model, topic): "overview",
+            executor.submit(run_video_agent, config.YOUTUBE_API_KEY, topic): "videos",
+            executor.submit(run_paper_agent, "", topic, 10): "papers",
+        }
+        for future in as_completed(futures):
+            key = futures[future]
+            try:
+                results[key] = future.result()
+                agent_log.append(f"✅ {key}_agent completed successfully")
+            except Exception as e:
+                results[key] = {"error": str(e)}
+                agent_log.append(f"❌ {key}_agent failed: {e}")
+
+    # Guard against a completely failed overview (nothing to aggregate)
+    if "content" not in results.get("overview", {}):
+        return {
+            "status": "error",
+            "reason": "Overview agent failed — cannot build report.",
+            "report": None,
+            "agent_log": agent_log,
+        }
+    
+    if "error" in results.get("overview", {}):
+        agent_log.append("⚠️ overview_agent rate-limited; loaded degraded fallback overview")
+
+    # --- Step 3: aggregate everything into one report ---------------------------
+    final_report = run_aggregator_agent(
+        gemini_model,
+        topic,
+        overview=results.get("overview", {}),
+        videos=results.get("videos", {}),
+        papers=results.get("papers", {}),
+    )
+    agent_log.append("✅ aggregator_agent merged all results")
+
+    # --- Step 4: safety gate on the FINAL report ---------------------------------
+    is_safe, reason = llm_safety_check(gemini_model, final_report)
+    if not is_safe:
+        agent_log.append(f"⛔ safety_gate blocked final report: {reason}")
+        return {
+            "status": "blocked",
+            "reason": reason,
+            "report": None,
+            "agent_log": agent_log,
+        }
+    
+    if reason and "Bypassed" in reason:
+        agent_log.append(f"⚠️ safety_gate bypassed: {reason}")
+    else:
+        agent_log.append("✅ safety_gate passed")
+
+    return {
+        "status": "success",
+        "reason": None,
+        "report": final_report,
+        "agent_log": agent_log,
+    }
+

backend/security.py

@@ -0,0 +1,74 @@
+"""
+security.py
+-----------
+Implements the two security controls we demonstrate for the capstone:
+
+1. NON-INTERACTIVE ACCESS
+   No agent ever fetches or renders raw third-party HTML. Every agent talks
+   only to structured, official APIs (Gemini, YouTube Data API, Semantic
+   Scholar, Google Books). This removes the indirect-prompt-injection
+   surface that comes from an agent reading arbitrary web pages.
+
+2. OUTPUT CONTENT SAFETY GATE (Governance / Logic Review)
+   Before the final report is returned to the user, the Aggregator Agent's
+   output passes through a lightweight safety classifier. If the topic or
+   generated content falls into a disallowed category (e.g. weapons,
+   self-harm instructions, hate content), the report is blocked and a safe
+   message is returned instead — mirroring the "Vibe Diff" / mandatory
+   logic-review idea from the security framework: nothing ships without a
+   check, even though the check here is automated rather than a human.
+"""
+
+import re
+
+# Minimal, fast keyword-based pre-filter. This runs before the LLM-based
+# check so obviously disallowed topics are caught without spending tokens.
+_BLOCKED_PATTERNS = [
+    r"\bmake\s+a\s+bomb\b",
+    r"\bsynthesiz(e|ing)\s+(nerve agent|sarin|ricin)\b",
+    r"\bhow to (kill|murder)\b",
+    r"\bchild\s+sexual\b",
+]
+
+
+def keyword_prefilter(topic: str) -> bool:
+    """Returns True if topic is safe to proceed, False if blocked outright."""
+    lowered = topic.lower()
+    for pattern in _BLOCKED_PATTERNS:
+        if re.search(pattern, lowered):
+            return False
+    return True
+
+
+def llm_safety_check(gemini_model, final_report_text: str) -> tuple[bool, str]:
+    """
+    Uses Gemini itself as a lightweight judge over the FINAL aggregated
+    report (not the raw topic) to catch anything that slipped through.
+
+    Returns (is_safe, reason).
+    """
+    prompt = (
+        "You are a content safety classifier. Read the study report below "
+        "and answer with exactly one word on the first line: SAFE or UNSAFE. "
+        "On the second line, give a one-sentence reason. "
+        "Mark UNSAFE only if the content provides actionable instructions "
+        "for violence, weapons, self-harm, or clearly illegal harmful acts. "
+        "Educational/academic framing of sensitive topics is SAFE.\n\n"
+        f"REPORT:\n{final_report_text[:4000]}"
+    )
+    try:
+        response = gemini_model["client"].models.generate_content(
+            model=gemini_model["model_name"], contents=prompt
+        )
+        text = response.text.strip()
+        first_line = text.splitlines()[0].strip().upper()
+        reason = text.splitlines()[1] if len(text.splitlines()) > 1 else ""
+        is_safe = first_line.startswith("SAFE")
+        return is_safe, reason
+    except Exception as e:
+        # If the check fails because of a 429 rate limit / quota exhaustion, bypass rather than block the whole application.
+        if "429" in str(e) or "RESOURCE_EXHAUSTED" in str(e) or "quota" in str(e).lower():
+            return True, f"Bypassed (API rate limit / quota exhausted)"
+        # Fail safe for other errors: if the classifier itself errors, don't silently ship unchecked content
+        return False, f"Safety check failed to run: {e}"
+

requirements.txt

@@ -0,0 +1,7 @@
+fastapi==0.115.0
+uvicorn[standard]==0.30.6
+google-genai==0.3.0
+requests==2.32.3
+python-dotenv==1.0.1
+pydantic==2.9.2
+streamlit==1.38.0