RAHULSR2806/devos-module2

1"""2DevOS Module 2 — Personal Knowledge Graph (RAG)3Ingest any content → embed → semantic search your own knowledge base4"""5 6import os7import sys8from typing import Optional9import httpx10from bs4 import BeautifulSoup11from fastapi import FastAPI, HTTPException12from fastapi.middleware.cors import CORSMiddleware13from pydantic import BaseModel14 15sys.path.append(os.path.join(os.path.dirname(__file__), ".."))16from shared.utils import get_supabase, groq_chat, embed_text, embed_texts, health_response17 18app = FastAPI(title="DevOS — Knowledge Graph", version="1.0.0")19 20app.add_middleware(21    CORSMiddleware,22    allow_origins=["*"],23    allow_methods=["*"],24    allow_headers=["*"],25)26 27 28# ─────────────────────────────────────────29# Models30# ─────────────────────────────────────────31class IngestRequest(BaseModel):32    content: str33    title: Optional[str] = None34    source_url: Optional[str] = None35    source_type: str = "note"  # article|code|note|video|paper36    tags: list[str] = []37 38 39class URLIngestRequest(BaseModel):40    url: str41    tags: list[str] = []42 43 44class SearchRequest(BaseModel):45    query: str46    limit: int = 1047    source_type: Optional[str] = None48    similarity_threshold: float = 0.349 50 51class AskRequest(BaseModel):52    question: str53    limit: int = 554 55 56# ─────────────────────────────────────────57# Content Processing58# ─────────────────────────────────────────59def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:60    """Split text into overlapping chunks for better retrieval"""61    words = text.split()62    chunks = []63    for i in range(0, len(words), chunk_size - overlap):64        chunk = " ".join(words[i : i + chunk_size])65        if chunk:66            chunks.append(chunk)67    return chunks68 69 70def extract_from_url(url: str) -> tuple[str, str]:71    """Fetch and extract clean text from a URL"""72    resp = httpx.get(url, timeout=15, follow_redirects=True)73    soup = BeautifulSoup(resp.text, "lxml")74 75    # Remove junk76    for tag in soup(["script", "style", "nav", "footer", "header", "aside"]):77        tag.decompose()78 79    title = soup.title.string if soup.title else url80    content = soup.get_text(separator=" ", strip=True)81    return title.strip(), content[:15000]82 83 84def generate_title_and_tags(content: str, existing_title: Optional[str]) -> tuple[str, list[str]]:85    prompt = f"""Given this content, extract a concise title (if not provided) and 3-5 relevant tags.86Content (first 500 chars): {content[:500]}87Existing title: {existing_title or 'None'}88 89Return JSON only:90{{"title": "...", "tags": ["tag1", "tag2", "tag3"]}}"""91 92    import json, re93    raw = groq_chat(94        messages=[{"role": "user", "content": prompt}],95        model="llama-3.3-70b-versatile",96        temperature=0.2,97        max_tokens=200,98    )99    clean = re.sub(r"```(?:json)?|```", "", raw).strip()100    try:101        data = json.loads(clean)102        return data.get("title", existing_title or "Untitled"), data.get("tags", [])103    except Exception:104        return existing_title or "Untitled", []105 106 107def store_knowledge_item(108    title: str,109    content: str,110    source_url: Optional[str],111    source_type: str,112    tags: list[str],113) -> list[str]:114    supabase = get_supabase()115    chunks = chunk_text(content)116    embeddings = embed_texts(chunks)117 118    ids = []119    parent_id = None120 121    for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):122        result = (123            supabase.table("knowledge_items")124            .insert(125                {126                    "title": title if i == 0 else f"{title} (part {i+1})",127                    "content": chunk,128                    "source_url": source_url,129                    "source_type": source_type,130                    "tags": tags,131                    "embedding": embedding,132                    "chunk_index": i,133                    "parent_id": parent_id,134                }135            )136            .execute()137        )138        item_id = result.data[0]["id"]139        if i == 0:140            parent_id = item_id141        ids.append(item_id)142 143    return ids144 145 146# ─────────────────────────────────────────147# Routes148# ─────────────────────────────────────────149@app.get("/health")150def health():151    return health_response("knowledge")152 153 154@app.post("/ingest")155def ingest_content(req: IngestRequest):156    """Ingest raw text/code/note into knowledge graph"""157    title = req.title158    tags = req.tags159 160    if not title or not tags:161        auto_title, auto_tags = generate_title_and_tags(req.content, req.title)162        title = title or auto_title163        tags = tags or auto_tags164 165    ids = store_knowledge_item(title, req.content, req.source_url, req.source_type, tags)166    return {"success": True, "chunks_stored": len(ids), "title": title, "tags": tags}167 168 169@app.post("/ingest/url")170def ingest_url(req: URLIngestRequest):171    """Fetch a URL and ingest its content"""172    try:173        title, content = extract_from_url(req.url)174    except Exception as e:175        raise HTTPException(status_code=400, detail=f"Could not fetch URL: {str(e)}")176 177    _, auto_tags = generate_title_and_tags(content, title)178    tags = req.tags or auto_tags179 180    ids = store_knowledge_item(title, content, req.url, "article", tags)181    return {"success": True, "chunks_stored": len(ids), "title": title, "tags": tags}182 183 184@app.post("/search")185def semantic_search(req: SearchRequest):186    """Semantic search across knowledge base"""187    supabase = get_supabase()188    query_embedding = embed_text(req.query)189 190    result = supabase.rpc(191        "search_knowledge",192        {193            "query_embedding": query_embedding,194            "similarity_threshold": req.similarity_threshold,195            "max_results": req.limit,196        },197    ).execute()198 199    results = result.data or []200    if req.source_type:201        results = [r for r in results if r.get("source_type") == req.source_type]202 203    return {"results": results, "total": len(results), "query": req.query}204 205 206@app.post("/ask")207def ask_knowledge_base(req: AskRequest):208    """RAG: search knowledge base then answer with Groq"""209    supabase = get_supabase()210    query_embedding = embed_text(req.question)211 212    result = supabase.rpc(213        "search_knowledge",214        {215            "query_embedding": query_embedding,216            "similarity_threshold": 0.25,217            "max_results": req.limit,218        },219    ).execute()220 221    context_items = result.data or []222    if not context_items:223        return {224            "answer": "I don't have any relevant knowledge about this topic yet. Try adding some articles or notes first!",225            "sources": [],226        }227 228    context = "\n\n---\n\n".join(229        [f"Source: {item['title']}\n{item['content']}" for item in context_items]230    )231 232    system = """You are a personal knowledge assistant. Answer questions using ONLY the provided context from the user's personal knowledge base. 233Be specific and cite which sources you're drawing from. If the context doesn't fully answer the question, say so."""234 235    prompt = f"""Context from my knowledge base:236{context}237 238Question: {req.question}"""239 240    answer = groq_chat(241        messages=[{"role": "user", "content": prompt}],242        model="llama-3.3-70b-versatile",243        system=system,244        temperature=0.2,245        max_tokens=1024,246    )247 248    sources = [249        {250            "title": item["title"],251            "source_url": item.get("source_url"),252            "similarity": round(item["similarity"], 3),253            "tags": item.get("tags", []),254        }255        for item in context_items256    ]257 258    return {"answer": answer, "sources": sources}259 260 261@app.get("/items")262def list_items(limit: int = 20, offset: int = 0, source_type: Optional[str] = None, tag: Optional[str] = None):263    supabase = get_supabase()264    query = (265        supabase.table("knowledge_items")266        .select("id, title, source_url, source_type, tags, chunk_index, created_at")267        .eq("chunk_index", 0)268        .order("created_at", desc=True)269        .range(offset, offset + limit - 1)270    )271    if source_type:272        query = query.eq("source_type", source_type)273    result = query.execute()274    return {"items": result.data, "total": len(result.data)}275 276 277@app.delete("/items/{item_id}")278def delete_item(item_id: str):279    supabase = get_supabase()280    supabase.table("knowledge_items").delete().or_(281        f"id.eq.{item_id},parent_id.eq.{item_id}"282    ).execute()283    return {"success": True}284 285 286@app.get("/stats")287def get_stats():288    supabase = get_supabase()289    items = supabase.table("knowledge_items").select("source_type, tags").eq("chunk_index", 0).execute()290    data = items.data or []291 292    type_counts: dict = {}293    all_tags: dict = {}294    for item in data:295        t = item.get("source_type", "note")296        type_counts[t] = type_counts.get(t, 0) + 1297        for tag in item.get("tags", []):298            all_tags[tag] = all_tags.get(tag, 0) + 1299 300    top_tags = sorted(all_tags.items(), key=lambda x: x[1], reverse=True)[:10]301    return {302        "total_documents": len(data),303        "by_type": type_counts,304        "top_tags": [{"tag": t, "count": c} for t, c in top_tags],305    }306