initial commit

app.py

@@ -0,0 +1,118 @@
+"""Streamlit semantic search app for CodeSeek AI."""
+
+from __future__ import annotations
+
+import os
+import sys
+import subprocess
+from pathlib import Path
+from typing import List
+
+import requests
+import streamlit as st
+
+from search_engine import SemanticSearchEngine
+
+
+# ================= CONFIG =================
+DEFAULT_MODEL = os.getenv("GITHUB_EMBEDDING_MODEL", "openai/text-embedding-3-small")
+API_VERSION = os.getenv("GITHUB_API_VERSION", "2026-03-10")
+BASE_URL = "https://models.github.ai"
+DATASET_PATH = Path("data/stackoverflow_sample_3000.json")
+
+
+# ================= ERRORS =================
+class GitHubModelsError(RuntimeError):
+    pass
+
+
+# ================= DATASET SETUP =================
+def ensure_dataset():
+    if not DATASET_PATH.exists():
+        with st.spinner("Preparing dataset (first run only)..."):
+            subprocess.run(
+                [sys.executable, "prepare_stackoverflow_sample.py"],
+                check=True
+            )
+
+
+# ================= ENGINE =================
+@st.cache_resource(show_spinner=False)
+def load_engine() -> SemanticSearchEngine:
+    return SemanticSearchEngine(DATASET_PATH)
+
+
+# ================= EMBEDDING =================
+def get_query_embedding(query: str) -> List[float]:
+    token = os.getenv("GITHUB_TOKEN")
+    org = os.getenv("GITHUB_ORG")
+
+    if not token:
+        raise GitHubModelsError("Missing GITHUB_TOKEN environment variable.")
+
+    endpoint = f"{BASE_URL}/inference/embeddings"
+    if org:
+        endpoint = f"{BASE_URL}/orgs/{org}/inference/embeddings"
+
+    headers = {
+        "Accept": "application/vnd.github+json",
+        "Authorization": f"Bearer {token}",
+        "X-GitHub-Api-Version": API_VERSION,
+        "Content-Type": "application/json",
+    }
+
+    payload = {
+        "model": DEFAULT_MODEL,
+        "input": query
+    }
+
+    response = requests.post(endpoint, headers=headers, json=payload, timeout=30)
+
+    if response.status_code >= 400:
+        raise GitHubModelsError(f"API Error {response.status_code}: {response.text[:300]}")
+
+    data = response.json()
+    return data["data"][0]["embedding"]
+
+
+# ================= MAIN APP =================
+def main():
+    st.set_page_config(page_title="CodeSeek AI", page_icon="🔎", layout="wide")
+
+    st.title("🔎 CodeSeek AI")
+    st.subheader("Semantic Programming Search")
+
+    # Ensure dataset exists
+    ensure_dataset()
+
+    query = st.text_area(
+        "Ask a programming question:",
+        placeholder="e.g. How to declare array in Python?",
+        height=120,
+    )
+
+    if not query.strip():
+        st.info("Enter a query to begin search.")
+        return
+
+    try:
+        with st.spinner("Searching..."):
+            engine = load_engine()
+            query_embedding = get_query_embedding(query.strip())
+            results = engine.search(query_embedding, top_k=5)
+
+    except Exception as e:
+        st.error(f"Search failed: {e}")
+        return
+
+    st.markdown("### Top Results")
+
+    for i, item in enumerate(results, start=1):
+        st.markdown(f"**{i}. {item['question']}**")
+        st.write(item["answer"])
+        st.caption(f"Similarity score: {item['score']:.4f}")
+        st.divider()
+
+
+if __name__ == "__main__":
+    main()

prepare_stackoverflow_sample.py

@@ -0,0 +1,35 @@
+"""Prepare a lightweight semantic search dataset from Hugging Face.
+
+Usage:
+    python prepare_stackoverflow_sample.py
+"""
+
+from pathlib import Path
+import json
+
+from datasets import load_dataset
+
+
+DATASET_ID = "MartinElMolon/stackoverflow_preguntas_con_embeddings"
+OUTPUT_PATH = Path("data/stackoverflow_sample_3000.json")
+SAMPLE_SIZE = 3000
+
+
+def main() -> None:
+    ds = load_dataset(DATASET_ID, split="train[:3000]")
+
+    if len(ds) < SAMPLE_SIZE:
+        raise ValueError(f"Dataset has only {len(ds)} rows; expected at least {SAMPLE_SIZE}.")
+
+    sampled = ds.shuffle(seed=42).select(range(SAMPLE_SIZE))
+    sampled = sampled.select_columns(["question", "answer", "embeddings"])
+
+    OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
+    with OUTPUT_PATH.open("w", encoding="utf-8") as f:
+        json.dump(sampled.to_list(), f, ensure_ascii=False)
+
+    print(f"Saved {len(sampled)} rows to {OUTPUT_PATH}")
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -1,3 +1,5 @@
-altair
-pandas
-streamlit
+streamlit
+faiss-cpu
+numpy
+requests
+datasets

search_engine.py

@@ -0,0 +1,102 @@
+"""Semantic vector search engine backed by FAISS.
+
+Expected dataset format (JSON array):
+[
+  {
+    "question": "...",
+    "answer": "...",
+    "embeddings": [0.1, 0.2, ...]
+  },
+  ...
+]
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import List, Dict, Any
+
+import faiss
+import numpy as np
+
+
+DEFAULT_DATASET_PATH = Path("data/stackoverflow_sample_3000.json")
+
+
+class SemanticSearchEngine:
+    """FAISS-based semantic search using cosine similarity via inner product."""
+
+    def __init__(self, dataset_path: str | Path = DEFAULT_DATASET_PATH) -> None:
+        self.dataset_path = Path(dataset_path)
+        self.metadata: List[Dict[str, str]] = []
+        self.embeddings: np.ndarray
+        self.index: faiss.IndexFlatIP
+        self._load_and_build()
+
+    def _load_and_build(self) -> None:
+        with self.dataset_path.open("r", encoding="utf-8") as f:
+            rows: List[Dict[str, Any]] = json.load(f)
+
+        if not isinstance(rows, list):
+            raise ValueError("Dataset must be a JSON array of objects.")
+        if not rows:
+            raise ValueError("Dataset is empty; expected at least one row.")
+
+        self.metadata = [
+            {
+                "question": row["question"],
+                "answer": row["answer"],
+            }
+            for row in rows
+        ]
+
+        embeddings = np.asarray([row["embeddings"] for row in rows], dtype=np.float32)
+        if embeddings.ndim != 2:
+            raise ValueError("Embeddings must be a 2D matrix [num_rows, dim].")
+
+        self.embeddings = self._normalize(embeddings)
+
+        dim = self.embeddings.shape[1]
+        self.index = faiss.IndexFlatIP(dim)
+        self.index.add(self.embeddings)
+
+    @staticmethod
+    def _normalize(vectors: np.ndarray) -> np.ndarray:
+        """L2-normalize vectors for cosine similarity search via inner product."""
+        vectors = np.asarray(vectors, dtype=np.float32)
+        norms = np.linalg.norm(vectors, axis=1, keepdims=True)
+        norms = np.where(norms == 0.0, 1.0, norms)
+        return vectors / norms
+
+    def search(self, query_embedding: List[float] | np.ndarray, top_k: int = 5) -> List[Dict[str, Any]]:
+        """Search nearest neighbors and return question/answer plus similarity score."""
+        if top_k <= 0:
+            raise ValueError("top_k must be greater than 0.")
+
+        query = np.asarray(query_embedding, dtype=np.float32).reshape(1, -1)
+        if query.shape[1] != self.embeddings.shape[1]:
+            raise ValueError(
+                f"Query dimension {query.shape[1]} does not match index dimension {self.embeddings.shape[1]}."
+            )
+
+        query = self._normalize(query)
+        scores, indices = self.index.search(query, min(top_k, len(self.metadata)))
+
+        results: List[Dict[str, Any]] = []
+        for score, idx in zip(scores[0], indices[0]):
+            item = self.metadata[int(idx)]
+            results.append(
+                {
+                    "question": item["question"],
+                    "answer": item["answer"],
+                    "score": float(score),
+                }
+            )
+        return results
+
+
+def search(query_embedding: List[float] | np.ndarray, top_k: int = 5) -> List[Dict[str, Any]]:
+    """Module-level convenience function using the default dataset path."""
+    engine = SemanticSearchEngine(DEFAULT_DATASET_PATH)
+    return engine.search(query_embedding=query_embedding, top_k=top_k)