update cursor

.gitignore

@@ -0,0 +1,13 @@
+.venv/
+__pycache__/
+.pytest_cache/
+.mypy_cache/
+.vector_index/
+.env
+*.parquet
+*.index
+*.ipynb_checkpoints/
+dist/
+build/
+*.egg-info/
+

Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.10-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PIP_NO_CACHE_DIR=1 \
+    HF_HUB_DISABLE_TELEMETRY=1
+
+WORKDIR /app
+
+COPY requirements.txt ./
+# Install Torch CPU wheels first to avoid heavy builds
+RUN pip install --upgrade pip && \
+    pip install --no-cache-dir --index-url https://download.pytorch.org/whl/cpu \
+      torch==2.4.1 torchvision==0.19.1 torchaudio==2.4.1 && \
+    pip install --no-cache-dir -r requirements.txt --default-timeout=100
+
+COPY . .
+
+EXPOSE 8000
+
+CMD ["python", "run.py"]
+

README.md

@@ -0,0 +1,221 @@
+## Feedback Analysis RAG Agent
+
+An end-to-end system for analyzing citizen feedback with Retrieval-Augmented Generation (RAG). It ingests `Feedback.csv`, creates multilingual embeddings, builds a FAISS vector index, and exposes a FastAPI API for semantic search, topic clustering, and sentiment summaries. Designed to run locally or in containers, and to be deployable to Runpod.
+
+### Features
+- Multilingual ingestion (Hebrew supported) from `Feedback.csv`
+- Preprocessing: optional normalization, language detection
+- Embeddings: Sentence-Transformers (multilingual) + FAISS
+- Retrieval: top-k semantic nearest neighbors with filters
+- Summarization: LLM (OpenAI) if configured; fallback to extractive summary
+  - Supports Gemini (preferred) or OpenAI when API keys are provided
+- Topics: k-means topic clustering over embeddings
+- Sentiment: multilingual transformer pipeline
+- FastAPI endpoints and a simple CLI
+
+### Project layout
+```
+app/
+  api.py
+  config.py
+  data_loader.py
+  embedding.py
+  preprocess.py
+  rag_service.py
+  sentiment.py
+  topics.py
+  vector_store.py
+run.py
+requirements.txt
+Dockerfile
+```
+
+### Quick start
+1) Python 3.10+
+2) Install:
+```
+python -m venv .venv && source .venv/bin/activate
+pip install -r requirements.txt
+```
+3) Environment (optional):
+```
+export OPENAI_API_KEY=sk-...
+export GEMINI_API_KEY=your_gemini_key
+```
+4) Run API:
+```
+python run.py
+```
+Open http://127.0.0.1:8000/docs
+
+5) CLI example:
+```
+python -m app.rag_service --query "שיפור טופס" --top_k 5
+```
+
+### Configuration
+Environment variables:
+- GEMINI_API_KEY: If set, RAG uses Gemini (preferred) for summaries
+- OPENAI_API_KEY: If set, RAG can use OpenAI as a fallback
+- EMBEDDING_MODEL: Sentence-Transformers model name (default: sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2)
+- VECTOR_INDEX_PATH: Path to persist FAISS index (default: ./.vector_index/faiss.index)
+
+### Notes
+- The first run will download models (embeddings, sentiment); ensure internet access.
+- The system reads from `Feedback.csv` in the repo root. Update `app/data_loader.py` if your schema differs.
+
+### Runpod
+- This repo includes a `Dockerfile`. Build and push the image; configure your Runpod template to run `python run.py` and expose port 8000.
+
+### Secrets hygiene
+- Do not commit real secrets. Use environment variables or a local `.env` file.
+- `.env` is gitignored by default via `.gitignore`.
+- Rotate any keys that were ever shared publicly.
+
+## Run on Runpod - Full guide
+
+### 1) Build and push the container
+- From project root:
+```
+docker build -t YOUR_DOCKERHUB_USER/feedback-rag:latest .
+docker login
+docker push YOUR_DOCKERHUB_USER/feedback-rag:latest
+```
+
+### 2) Prepare environment variables (no secrets in git)
+- You will set secrets within Runpod, not in code:
+  - Required:
+    - `GEMINI_API_KEY` = your Gemini key
+  - Optional:
+    - `OPENAI_API_KEY` = OpenAI fallback key
+    - `CSV_PATH` = path to your CSV if not the default `Feedback.csv`
+    - `VECTOR_INDEX_PATH` and `VECTOR_METADATA_PATH` if you change mount/paths
+
+### 3) Create a Runpod Template (Serverless HTTP recommended)
+- In Runpod Console → Templates → Create Template
+- Fields:
+  - Container Image: `YOUR_DOCKERHUB_USER/feedback-rag:latest`  (if you want you can use mine: `galbendavids/feedback-rag:latest`)
+  - Container Port: `8000`
+  - Command: `python run.py`
+  - Environment Variables:
+    - `GEMINI_API_KEY=your_key`
+    - (optional) `OPENAI_API_KEY=sk-...`
+    - (optional) `CSV_PATH=/workspace/Feedback.csv`
+    - (optional) `VECTOR_INDEX_PATH=/workspace/.vector_index/faiss.index`
+    - (optional) `VECTOR_METADATA_PATH=/workspace/.vector_index/meta.parquet`
+- Volumes (recommended to persist the FAISS index):
+  - Create a volume, mount it at `/workspace/.vector_index`
+  - Make sure your `VECTOR_*` env vars point to that mount path if changed
+
+### 4) Deploy a Serverless Endpoint
+- Create Endpoint from the template (Serverless)
+- Choose region and CPU (CPU is sufficient)
+- Wait until status is Running and an endpoint URL is provided
+
+### 5) Upload or point to your CSV
+- Option A (bundled): Keep `Feedback.csv` in the image (already in repo root)
+- Option B (mounted): Upload to a mounted volume and set `CSV_PATH` accordingly
+
+### 6) First-time ingestion (build the vector index)
+- Trigger ingestion once to build and persist the FAISS index:
+```
+curl -X POST {YOUR_ENDPOINT_URL}/ingest
+```
+- On first run, models download and embeddings are computed; allow a few minutes
+- The index will be stored under `.vector_index` (persist if using a volume)
+
+### 7) Test the API
+- Health:
+```
+curl -s {YOUR_ENDPOINT_URL}/health
+```
+- Query:
+```
+curl -X POST {YOUR_ENDPOINT_URL}/query \
+  -H "Content-Type: application/json" \
+  -d '{"query":"שיפור טופס", "top_k": 5}' \
+  {YOUR_ENDPOINT_URL}/query
+```
+- Topics:
+```
+curl -s "{YOUR_ENDPOINT_URL}/topics?num_topics=8"
+```
+- Sentiment (first N rows):
+```
+curl -s "{YOUR_ENDPOINT_URL}/sentiment?limit=100"
+```
+- Interactive docs (Swagger UI):
+  - Open `{YOUR_ENDPOINT_URL}/docs` in your browser
+
+### 8) Using Dedicated Pods (alternative)
+- Launch a Dedicated Pod from the template
+- Ensure command `python run.py` and port `8000`
+- Use the Pod’s public endpoint to access `/health`, `/ingest`, `/query`, etc.
+
+### 9) Troubleshooting
+- 404/connection:
+  - Endpoint not Running yet or wrong port; port must be `8000`
+- Slow initial response:
+  - First-time model downloads are expected; subsequent calls are faster
+- No/few results:
+  - Ensure you POSTed `/ingest` first and that your CSV has the `Text` column
+- Index not persisted:
+  - Mount a volume at `/workspace/.vector_index` and set `VECTOR_*` paths
+
+### 10) Optional: Pre-cache models to speed cold starts
+- You can pre-bake model weights in the image by adding to your `Dockerfile`:
+```
+# Optional: pre-download models during build to reduce cold start time
+RUN python -c "from sentence_transformers import SentenceTransformer; SentenceTransformer('sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2')"
+RUN python -c "from transformers import pipeline; pipeline('sentiment-analysis', model='cardiffnlp/twitter-xlm-roberta-base-sentiment')"
+```
+- Rebuild and push the image after adding these lines.
+
+## Offline precompute (embed the DB locally for fast startup)
+
+If you want the API to start fast on Runpod without running `/ingest` there, precompute the vector index locally:
+
+1) Create venv and install deps:
+```
+python -m venv .venv && source .venv/bin/activate
+pip install -r requirements.txt
+```
+2) Ensure `Feedback.csv` exists at repo root (or set `CSV_PATH`).
+3) Run the offline precompute script:
+```
+python scripts/precompute_index.py
+```
+This writes:
+- `.vector_index/faiss.index`
+- `.vector_index/meta.parquet`
+
+4) Option A: Commit the index (makes startup fastest)
+- By default `.vector_index/` is in `.gitignore`. To commit it, you can temporarily remove that entry and run:
+```
+git add .vector_index/faiss.index .vector_index/meta.parquet
+git commit -m "Add precomputed FAISS index"
+git push
+```
+(Note: repo size will increase; acceptable for small indices.)
+
+5) Option B: Keep index uncommitted; mount it on Runpod
+- Upload the `.vector_index/` folder to a Runpod volume mounted at `/workspace/.vector_index`
+- Set env vars if you changed paths:
+  - `VECTOR_INDEX_PATH=/workspace/.vector_index/faiss.index`
+  - `VECTOR_METADATA_PATH=/workspace/.vector_index/meta.parquet`
+
+With either option, the API will be immediately queryable without calling `/ingest`.
+
+### When your data changes
+- If you update `Feedback.csv` (or change `CSV_PATH` to a new dataset), you must rerun:
+```
+uv run -m scripts.precompute_index
+```
+- Then redeploy (bake new files into the image or upload to your Runpod volume) so the server uses the fresh index.
+
+### Adding new feedback entries
+- You can add rows to `Feedback.csv` and either:
+  - Rebuild the entire index (simple, safest):
+    - `uv run -m scripts.precompute_index`
+  - Or implement an incremental append (advanced): embed only the new rows with `EmbeddingModel.encode(...)`, call `FaissVectorStore.load(...)`, then `store.add(new_vectors, new_metadata)` and `store.save(...)`. This keeps the same architecture and avoids re-embedding all previous data.
+

app/__init__.py

@@ -0,0 +1,2 @@
+# Makes `app` a package so imports like `from app.rag_service import RAGService` work.
+

app/api.py

@@ -0,0 +1,177 @@
+from __future__ import annotations
+
+from typing import List, Optional, Dict, Any
+
+import numpy as np
+import pandas as pd
+from fastapi import FastAPI, Query
+from pydantic import BaseModel
+
+from .config import settings
+from .data_loader import load_feedback
+from .embedding import EmbeddingModel
+from .rag_service import RAGService
+from .sentiment import analyze_sentiments
+from .topics import kmeans_topics
+from .vector_store import FaissVectorStore
+
+
+app = FastAPI(title="Feedback Analysis RAG Agent", version="1.0.0", default_response_class=None)
+svc = RAGService()
+embedder = svc.embedder
+
+
+class QueryRequest(BaseModel):
+    query: str
+    top_k: int = 5
+
+
+class QueryResponse(BaseModel):
+    query: str
+    summary: Optional[str]
+    results: List[Dict[str, Any]]
+
+
+@app.get("/health")
+def health() -> Dict[str, str]:
+    return {"status": "ok"}
+
+
+@app.post("/ingest")
+def ingest() -> Dict[str, Any]:
+    """Build the vector index from Feedback.csv"""
+    try:
+        svc.ingest()
+        return {"status": "ingested", "message": "Vector index built successfully"}
+    except FileNotFoundError as e:
+        return {"status": "error", "message": f"CSV file not found: {str(e)}"}
+    except Exception as e:
+        return {"status": "error", "message": f"Ingestion failed: {str(e)}"}
+
+
+@app.post("/query", response_model=QueryResponse)
+def query(req: QueryRequest) -> QueryResponse:
+    """Free-form question answering over feedback data."""
+    try:
+        out = svc.query(req.query, top_k=req.top_k)
+        return QueryResponse(
+            query=out.query,
+            summary=out.summary,
+            results=[
+                {
+                    "score": r.score,
+                    "service": r.row.get(settings.service_column, ""),
+                    "level": r.row.get(settings.level_column, ""),
+                    "text": r.row.get(settings.text_column, ""),
+                }
+                for r in out.results
+            ],
+        )
+    except FileNotFoundError:
+        return QueryResponse(
+            query=req.query,
+            summary="Error: Vector index not found. Please run /ingest first.",
+            results=[]
+        )
+    except Exception as e:
+        return QueryResponse(
+            query=req.query,
+            summary=f"Error: {str(e)}",
+            results=[]
+        )
+
+
+@app.get("/topics")
+def topics(num_topics: int = Query(5, ge=2, le=50)) -> Dict[str, Any]:
+    """Extract main topics from feedback. Returns topics with summaries."""
+    try:
+        # Load embeddings from store
+        store = FaissVectorStore.load(settings.vector_index_path, settings.vector_metadata_path)
+        # FAISS does not expose vectors, so recompute for this endpoint
+        df = load_feedback()
+        texts = df[settings.text_column].astype(str).tolist()
+        if not texts:
+            return {"num_topics": 0, "topics": {}, "error": "No feedback data found"}
+        
+        embeddings = embedder.encode(texts)
+        res = kmeans_topics(embeddings, num_topics=num_topics)
+        
+        # Group texts by topic
+        topics_out: Dict[int, List[str]] = {}
+        for label, text in zip(res.labels, texts):
+            topics_out.setdefault(int(label), []).append(text)
+        
+        # Generate topic names/summaries using LLM if available
+        topic_summaries: Dict[int, str] = {}
+        for topic_id, topic_texts in topics_out.items():
+            # Take sample texts for summary
+            sample_texts = topic_texts[:10] if len(topic_texts) > 10 else topic_texts
+            sample_str = "\n".join(f"- {t[:200]}" for t in sample_texts[:5])
+            
+            prompt = (
+                "Based on the following citizen feedback examples, provide a short topic name (2-4 words) "
+                "in Hebrew that describes what users are talking about. "
+                "Return ONLY the topic name, nothing else.\n\n"
+                f"Examples:\n{sample_str}\n\nTopic name:"
+            )
+            
+            topic_name = f"נושא {topic_id + 1}"  # Default fallback
+            
+            # Try Gemini first
+            if settings.gemini_api_key:
+                try:
+                    import google.generativeai as genai
+                    genai.configure(api_key=settings.gemini_api_key)
+                    model = genai.GenerativeModel("gemini-1.5-flash")
+                    resp = model.generate_content(prompt)
+                    text = getattr(resp, "text", None)
+                    if isinstance(text, str) and text.strip():
+                        topic_name = text.strip()
+                except Exception:
+                    pass
+            
+            # Fallback to OpenAI
+            if topic_name.startswith("נושא") and settings.openai_api_key:
+                try:
+                    from openai import OpenAI
+                    client = OpenAI(api_key=settings.openai_api_key)
+                    resp = client.chat.completions.create(
+                        model="gpt-4o-mini",
+                        messages=[{"role": "user", "content": prompt}],
+                        temperature=0.3,
+                        max_tokens=20,
+                    )
+                    if resp.choices[0].message.content:
+                        topic_name = resp.choices[0].message.content.strip()
+                except Exception:
+                    pass
+            
+            topic_summaries[topic_id] = topic_name
+        
+        # Format response with topic names
+        formatted_topics: Dict[str, Any] = {}
+        for topic_id, topic_texts in topics_out.items():
+            formatted_topics[str(topic_id)] = {
+                "name": topic_summaries.get(topic_id, f"נושא {topic_id + 1}"),
+                "count": len(topic_texts),
+                "examples": topic_texts[:5]  # First 5 examples
+            }
+        
+        return {
+            "num_topics": num_topics,
+            "topics": formatted_topics,
+            "total_feedback": len(texts)
+        }
+    except FileNotFoundError:
+        return {"error": "Vector index not found. Please run /ingest first.", "num_topics": 0, "topics": {}}
+    except Exception as e:
+        return {"error": str(e), "num_topics": 0, "topics": {}}
+
+
+@app.get("/sentiment")
+def sentiment(limit: int = Query(100, ge=1, le=2000)) -> Dict[str, Any]:
+    df = load_feedback().head(limit)
+    texts = df[settings.text_column].astype(str).tolist()
+    out = analyze_sentiments(texts)
+    return {"count": len(out), "results": out}
+

app/config.py

@@ -0,0 +1,27 @@
+import os
+from dataclasses import dataclass
+from dotenv import load_dotenv  # type: ignore
+
+
+# Load .env if present (kept out of git via .gitignore)
+load_dotenv(override=False)
+
+
+@dataclass
+class Settings:
+    openai_api_key: str | None = os.getenv("OPENAI_API_KEY")
+    gemini_api_key: str | None = os.getenv("GEMINI_API_KEY")
+    embedding_model_name: str = os.getenv(
+        "EMBEDDING_MODEL",
+        "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
+    )
+    vector_index_path: str = os.getenv("VECTOR_INDEX_PATH", ".vector_index/faiss.index")
+    vector_metadata_path: str = os.getenv("VECTOR_METADATA_PATH", ".vector_index/meta.parquet")
+    csv_path: str = os.getenv("CSV_PATH", "Feedback.csv")
+    text_column: str = os.getenv("TEXT_COLUMN", "Text")
+    service_column: str = os.getenv("SERVICE_COLUMN", "ServiceName")
+    level_column: str = os.getenv("LEVEL_COLUMN", "Level")
+
+
+settings = Settings()
+

app/data_loader.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+import pandas as pd
+from .config import settings
+
+
+def load_feedback(csv_path: str | None = None) -> pd.DataFrame:
+    path = csv_path or settings.csv_path
+    df = pd.read_csv(path)
+    # Basic normalization of expected columns if present
+    expected = ["ID", "ServiceName", "Level", "Text"]
+    missing = [c for c in expected if c not in df.columns]
+    if missing:
+        raise ValueError(f"Missing expected columns in CSV: {missing}")
+    # Drop rows with empty text
+    df = df[df["Text"].astype(str).str.strip().ne("")].copy()
+    df.reset_index(drop=True, inplace=True)
+    return df
+

app/embedding.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from typing import Iterable, List
+
+import numpy as np
+from sentence_transformers import SentenceTransformer  # type: ignore
+
+from .config import settings
+
+
+class EmbeddingModel:
+    def __init__(self, model_name: str | None = None) -> None:
+        self.model_name = model_name or settings.embedding_model_name
+        self.model = SentenceTransformer(self.model_name)
+
+    def encode(self, texts: Iterable[str], batch_size: int = 32) -> np.ndarray:
+        embeddings = self.model.encode(
+            list(texts),
+            batch_size=batch_size,
+            show_progress_bar=True,
+            convert_to_numpy=True,
+            normalize_embeddings=True,
+        )
+        return embeddings
+
+    def encode_single(self, text: str) -> np.ndarray:
+        return self.encode([text])[0]
+

app/preprocess.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+from langdetect import detect, DetectorFactory  # type: ignore
+
+DetectorFactory.seed = 42
+
+
+def detect_language(text: str) -> str:
+    try:
+        return detect(text)
+    except Exception:
+        return "unknown"
+
+
+def normalize_text(text: str) -> str:
+    # Minimal normalization; keep non-latin scripts (Hebrew)
+    return " ".join(str(text).split())
+
+
+def preprocess_text(text: str) -> str:
+    return normalize_text(text)
+

app/rag_service.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+import argparse
+from dataclasses import dataclass
+from typing import List, Optional, Dict
+
+import numpy as np
+import pandas as pd
+
+from .config import settings
+from .data_loader import load_feedback
+from .embedding import EmbeddingModel
+from .preprocess import preprocess_text
+from .vector_store import FaissVectorStore, SearchResult
+
+
+try:
+    from openai import OpenAI  # type: ignore
+except Exception:  # pragma: no cover - optional
+    OpenAI = None  # type: ignore
+
+try:
+    import google.generativeai as genai  # type: ignore
+except Exception:  # pragma: no cover - optional
+    genai = None  # type: ignore
+
+
+@dataclass
+class RetrievalOutput:
+    query: str
+    results: List[SearchResult]
+    summary: Optional[str]
+
+
+class RAGService:
+    def __init__(self) -> None:
+        self.embedder = EmbeddingModel()
+        self.store: Optional[FaissVectorStore] = None
+
+    def ingest(self, df: Optional[pd.DataFrame] = None) -> None:
+        data = df if df is not None else load_feedback()
+        texts = [preprocess_text(t) for t in data[settings.text_column].astype(str).tolist()]
+        vectors = self.embedder.encode(texts)
+
+        store = FaissVectorStore(dim=vectors.shape[1])
+        store.add(vectors.astype(np.float32), data[[settings.text_column, settings.service_column, settings.level_column]])
+        store.save(settings.vector_index_path, settings.vector_metadata_path)
+        self.store = store
+
+    def _ensure_store(self) -> None:
+        if self.store is None:
+            import os
+            if not os.path.exists(settings.vector_index_path):
+                raise FileNotFoundError(
+                    f"Vector index not found at {settings.vector_index_path}. "
+                    "Please run /ingest endpoint first or precompute the index."
+                )
+            self.store = FaissVectorStore.load(settings.vector_index_path, settings.vector_metadata_path)
+
+    def retrieve(self, query: str, top_k: int = 5) -> List[SearchResult]:
+        self._ensure_store()
+        assert self.store is not None
+        q_vec = self.embedder.encode_single(preprocess_text(query))
+        results = self.store.search(q_vec, top_k=top_k)
+        return results
+
+    def summarize(self, query: str, contexts: List[str]) -> Optional[str]:
+        if not contexts:
+            return None
+        joined = "\n".join(f"- {c}" for c in contexts[:10])
+        # Detect if query is in Hebrew
+        is_hebrew = any('\u0590' <= char <= '\u05FF' for char in query)
+        lang_instruction = "ענה בעברית" if is_hebrew else "Answer in the language of the query"
+        
+        prompt = (
+            f"You are a government digital services assistant. Based on the following citizen feedback snippets, "
+            f"write a concise summary (max 100 words) highlighting key issues and suggestions. "
+            f"{lang_instruction}.\n\n"
+            f"Query:\n{query}\n\nFeedback:\n{joined}\n\nSummary:"
+        )
+        # Prefer Gemini if configured
+        if settings.gemini_api_key and genai is not None:
+            try:
+                genai.configure(api_key=settings.gemini_api_key)
+                model = genai.GenerativeModel("gemini-1.5-flash")
+                resp = model.generate_content(prompt)
+                text = getattr(resp, "text", None)
+                if isinstance(text, str) and text.strip():
+                    return text.strip()
+            except Exception:
+                pass
+        # Fallback to OpenAI if available
+        if settings.openai_api_key and OpenAI is not None:
+            client = OpenAI(api_key=settings.openai_api_key)
+            try:
+                resp = client.chat.completions.create(
+                    model="gpt-4o-mini",
+                    messages=[{"role": "user", "content": prompt}],
+                    temperature=0.2,
+                    max_tokens=200,
+                )
+                return resp.choices[0].message.content
+            except Exception:
+                pass
+        # Fallback: simple extractive "summary"
+        return " ".join(contexts[:3])
+
+    def query(self, query: str, top_k: int = 5) -> RetrievalOutput:
+        results = self.retrieve(query, top_k=top_k)
+        contexts = [r.row[settings.text_column] for r in results]
+        summary = self.summarize(query, contexts)
+        return RetrievalOutput(query=query, results=results, summary=summary)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--ingest", action="store_true", help="Ingest CSV and build index")
+    parser.add_argument("--query", type=str, default=None, help="Run a semantic query")
+    parser.add_argument("--top_k", type=int, default=5, help="Top K results")
+    args = parser.parse_args()
+
+    svc = RAGService()
+    if args.ingest:
+        svc.ingest()
+        print("Ingest completed.")
+    if args.query:
+        out = svc.query(args.query, top_k=args.top_k)
+        print("Summary:", out.summary)
+        for r in out.results:
+            print(f"[{r.score:.3f}] {r.row.get('ServiceName','')} | {r.row.get('Text','')[:200]}")
+
+
+if __name__ == "__main__":
+    main()
+

app/sentiment.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from functools import lru_cache
+from typing import List, Dict
+
+from transformers import pipeline  # type: ignore
+
+
+@lru_cache(maxsize=1)
+def get_sentiment_pipeline():
+    # Multilingual sentiment model
+    return pipeline("sentiment-analysis", model="cardiffnlp/twitter-xlm-roberta-base-sentiment")
+
+
+def analyze_sentiments(texts: List[str]) -> List[Dict[str, float | str]]:
+    clf = get_sentiment_pipeline()
+    outputs = clf(texts, truncation=True)
+    results: List[Dict[str, float | str]] = []
+    for out in outputs:
+        label = out.get("label", "")
+        score = float(out.get("score", 0.0))
+        results.append({"label": label, "score": score})
+    return results
+

app/topics.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import List, Dict
+
+import numpy as np
+from sklearn.cluster import KMeans  # type: ignore
+
+
+@dataclass
+class TopicResult:
+    labels: List[int]
+    centroids: np.ndarray
+
+
+def kmeans_topics(embeddings: np.ndarray, num_topics: int = 8, seed: int = 42) -> TopicResult:
+    if len(embeddings) == 0:
+        return TopicResult(labels=[], centroids=np.empty((0, embeddings.shape[1])))
+    km = KMeans(n_clusters=num_topics, random_state=seed, n_init="auto")
+    labels = km.fit_predict(embeddings)
+    return TopicResult(labels=list(map(int, labels)), centroids=km.cluster_centers_) 
+

app/vector_store.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from typing import List, Tuple, Optional
+
+import faiss  # type: ignore
+import numpy as np
+import pandas as pd
+
+from .config import settings
+
+
+@dataclass
+class SearchResult:
+    index: int
+    score: float
+    row: pd.Series
+
+
+class FaissVectorStore:
+    def __init__(self, dim: int) -> None:
+        self.dim = dim
+        self.index = faiss.IndexFlatIP(dim)
+        self.metadata: Optional[pd.DataFrame] = None
+
+    def add(self, vectors: np.ndarray, metadata: pd.DataFrame) -> None:
+        if vectors.dtype != np.float32:
+            vectors = vectors.astype(np.float32)
+        if self.metadata is None:
+            self.metadata = metadata.reset_index(drop=True)
+        else:
+            self.metadata = pd.concat([self.metadata, metadata], ignore_index=True)
+        self.index.add(vectors)
+
+    def search(self, query_vector: np.ndarray, top_k: int = 5) -> List[SearchResult]:
+        q = query_vector.astype(np.float32).reshape(1, -1)
+        scores, idxs = self.index.search(q, top_k)
+        results: List[SearchResult] = []
+        for score, idx in zip(scores[0], idxs[0]):
+            if idx < 0 or self.metadata is None:
+                continue
+            results.append(SearchResult(index=int(idx), score=float(score), row=self.metadata.iloc[int(idx)]))
+        return results
+
+    def save(self, vector_path: str, meta_path: str) -> None:
+        os.makedirs(os.path.dirname(vector_path), exist_ok=True)
+        faiss.write_index(self.index, vector_path)
+        if self.metadata is not None:
+            self.metadata.to_parquet(meta_path, index=False)
+
+    @classmethod
+    def load(cls, vector_path: str, meta_path: str) -> "FaissVectorStore":
+        index = faiss.read_index(vector_path)
+        dim = index.d
+        store = cls(dim=dim)
+        store.index = index
+        if os.path.exists(meta_path):
+            store.metadata = pd.read_parquet(meta_path)
+        return store
+

requirements.txt

@@ -0,0 +1,17 @@
+fastapi==0.115.5
+uvicorn[standard]==0.32.0
+pandas==2.2.3
+numpy==1.26.4
+scikit-learn==1.5.2
+faiss-cpu==1.8.0.post1
+sentence-transformers==3.1.1
+transformers==4.45.2
+torch==2.4.1
+langdetect==1.0.9
+openai==1.52.2
+python-dotenv==1.0.1
+pydantic==2.9.2
+orjson==3.10.7
+google-generativeai==0.6.0
+pyarrow==14.0.2
+

run.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+import uvicorn  # type: ignore
+
+
+def main() -> None:
+    uvicorn.run("app.api:app", host="0.0.0.0", port=8000, reload=False)
+
+
+if __name__ == "__main__":
+    main()
+

scripts/__init__.py

@@ -0,0 +1,2 @@
+# Optional: mark `scripts` as a package to allow module-style execution.
+

scripts/precompute_index.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from app.rag_service import RAGService
+from app.config import settings
+
+
+def main() -> None:
+    out_dir = Path(settings.vector_index_path).parent
+    out_dir.mkdir(parents=True, exist_ok=True)
+    svc = RAGService()
+    svc.ingest()
+    print(f"Index written to: {settings.vector_index_path}")
+    print(f"Metadata written to: {settings.vector_metadata_path}")
+
+
+if __name__ == "__main__":
+    main()
+