feat: implement backend ingestion pipeline with Celery workers, Qdrant integration, and session management

backend/clean_db.py

@@ -2,7 +2,8 @@ import asyncio
 from src.database import engine
 from sqlalchemy import text
 from src.config import get_settings
-from pinecone import Pinecone, ServerlessSpec
+from qdrant_client import QdrantClient
+from qdrant_client.models import Distance, VectorParams, PayloadSchemaType
 
 async def main():
     async with engine.begin() as conn:
@@ -11,22 +12,27 @@ async def main():
         await conn.execute(text('CREATE SCHEMA public'))
         print('Postgres schema wiped.')
 
+    import qdrant_client
     settings = get_settings()
     try:
-        pc = Pinecone(api_key=settings.pinecone_api_key)
-        existing = [idx.name for idx in pc.list_indexes()]
-        if settings.index_name in existing:
-            pc.delete_index(settings.index_name)
+        q = QdrantClient(url=settings.qdrant_url, api_key=settings.qdrant_api_key)
+        q.delete_collection(settings.collection_name)
+        q.create_collection(settings.collection_name, vectors_config=VectorParams(size=384, distance=Distance.COSINE))
         
-        pc.create_index(
-            name=settings.index_name,
-            dimension=settings.vector_size,
-            metric='cosine',
-            spec=ServerlessSpec(cloud='aws', region='us-east-1')
+        # Reinject indices required natively by the pipeline
+        q.create_payload_index(
+            collection_name=settings.collection_name,
+            field_name="session_id",
+            field_schema=PayloadSchemaType.KEYWORD
         )
-        print('Pinecone index wiped and recreated.')
+        q.create_payload_index(
+            collection_name=settings.collection_name,
+            field_name="years_of_experience",
+            field_schema=PayloadSchemaType.FLOAT
+        )
+        print('Qdrant collection wiped and re-indexed.')
     except Exception as e:
-        print('Pinecone error:', e)
+        print('Qdrant error:', e)
 
     print("\n------------------------------")
     print("Database is completely purged but empty.")

backend/main.py

@@ -1,12 +1,11 @@
 import os
 import logging
-import json
-import redis.asyncio as redis
 from contextlib import asynccontextmanager
 from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.staticfiles import StaticFiles
-from pinecone import Pinecone, ServerlessSpec
+from qdrant_client import QdrantClient
+from qdrant_client.models import Distance, VectorParams, PayloadSchemaType
 
 from src.config import get_settings
 from src.models import JobDescription, Candidate, MatchResult, Session
@@ -15,48 +14,52 @@ from src.routers import jds, candidates, matching, sessions
 logger = logging.getLogger(__name__)
 settings = get_settings()
 
-_pc_client: Pinecone | None = None
-_pc_ready: bool = False
-_redis_client: redis.Redis | None = None
+_qdrant_client: QdrantClient | None = None
+_qdrant_ready: bool = False
 
 
-def get_pinecone() -> Pinecone:
-    return _pc_client
-
-
-def get_redis() -> redis.Redis:
-    return _redis_client
+def get_qdrant() -> QdrantClient:
+    return _qdrant_client
 
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    global _pc_client, _pc_ready, _redis_client
-    _pc_client = Pinecone(api_key=settings.pinecone_api_key)
-    _redis_client = redis.from_url(settings.redis_url, decode_responses=False)
+    global _qdrant_client, _qdrant_ready
+    _qdrant_client = QdrantClient(url=settings.qdrant_url, api_key=settings.qdrant_api_key)
 
     try:
-        existing = [idx.name for idx in _pc_client.list_indexes()]
-        if settings.index_name not in existing:
-            _pc_client.create_index(
-                name=settings.index_name,
-                dimension=settings.vector_size,
-                metric='cosine',
-                spec=ServerlessSpec(cloud='aws', region='us-east-1')
+        existing = [c.name for c in _qdrant_client.get_collections().collections]
+        if settings.collection_name not in existing:
+            _qdrant_client.create_collection(
+                collection_name=settings.collection_name,
+                vectors_config=VectorParams(size=settings.vector_size, distance=Distance.COSINE),
+            )
+            # Create indexing for the session_id to allow fast filtering
+            _qdrant_client.create_payload_index(
+                collection_name=settings.collection_name,
+                field_name="session_id",
+                field_schema=PayloadSchemaType.UUID,
+            )
+            # Create indexing for years_of_experience for range filtering
+            _qdrant_client.create_payload_index(
+                collection_name=settings.collection_name,
+                field_name="years_of_experience",
+                field_schema=PayloadSchemaType.FLOAT,
             )
-        _pc_ready = True
-        logger.info("Pinecone connected — index '%s' ready", settings.index_name)
+        _qdrant_ready = True
+        logger.info("Qdrant connected — collection '%s' ready", settings.collection_name)
     except Exception as exc:
-        _pc_ready = False
+        _qdrant_ready = False
         logger.warning(
-            "Pinecone unavailable at startup (%s).",
+            "Qdrant unavailable at startup (%s). "
+            "The API will start but vector search will fail until Qdrant is reachable.",
             exc,
         )
 
-    app.state.pinecone = _pc_client
-    app.state.pinecone_ready = _pc_ready
-    app.state.redis = _redis_client
+    app.state.qdrant = _qdrant_client
+    app.state.qdrant_ready = _qdrant_ready
     yield
-    await _redis_client.close()
+    _qdrant_client.close()
 
 
 app = FastAPI(
@@ -82,13 +85,11 @@ app.include_router(matching.router, prefix="/api/match", tags=["Matching"])
 
 @app.get("/health")
 async def health(request: "Request"):
-    pc_ok = getattr(request.app.state, "pinecone_ready", False)
-    redis_conn = getattr(request.app.state, "redis", None)
+    qdrant_ok = getattr(request.app.state, "qdrant_ready", False)
     return {
         "status": "ok",
         "version": "1.0.0",
-        "pinecone": "connected" if pc_ok else "unavailable",
-        "redis": "ok" if redis_conn else "missing",
+        "qdrant": "connected" if qdrant_ok else "unavailable",
     }
 
 

backend/requirements.txt

@@ -4,6 +4,7 @@ sqlalchemy[asyncio]==2.0.36
 asyncpg==0.30.0
 psycopg[binary]==3.2.3
 alembic==1.13.3
+qdrant-client==1.11.0
 celery[redis]==5.4.0
 redis==5.1.0
 sentence-transformers==3.3.0
@@ -17,4 +18,5 @@ numpy==1.26.4
 python-dotenv==1.0.1
 httpx==0.27.2
 aiofiles==24.1.0
+pinecone-client
 pinecone

backend/src/config.py

@@ -7,13 +7,14 @@ class Settings(BaseSettings):
     model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
 
     database_url: str
-    pinecone_api_key: str
+    qdrant_url: str
+    qdrant_api_key: str
     redis_url: str
     groq_api_key: str
     groq_model: str = "llama-3.3-70b-versatile"
     embedding_model: str = "BAAI/bge-small-en-v1.5"
     reranker_model: str = "BAAI/bge-reranker-v2-m3"
-    index_name: str = "candidates-v1"
+    collection_name: str = "candidates_v1"
     vector_size: int = 384
 
     @property

backend/src/matching/stage1.py

@@ -1,5 +1,6 @@
 from typing import Any
-from pinecone import Pinecone
+from qdrant_client import QdrantClient
+from qdrant_client.models import Filter, FieldCondition, MatchValue, Range
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy import select
 
@@ -15,19 +16,21 @@ DEFAULT_WEIGHTS = {
 }
 
 
-def _build_pinecone_filter(jd: dict, session_id: str | None) -> dict | None:
-    conditions = {}
+def _build_qdrant_filter(jd: dict, session_id: str | None) -> Filter | None:
+    conditions = []
     if session_id:
-        conditions["session_id"] = session_id
+        conditions.append(FieldCondition(key="session_id", match=MatchValue(value=session_id)))
     if jd.get("min_yoe") is not None:
-        conditions["years_of_experience"] = {"$gte": float(max(0, jd["min_yoe"] - 2))}
-    return conditions if conditions else None
+        conditions.append(FieldCondition(key="years_of_experience", range=Range(gte=max(0, jd["min_yoe"] - 2))))
+    if not conditions:
+        return None
+    return Filter(must=conditions)
 
 
 async def stage1_retrieve(
     jd: dict,
     db: AsyncSession,
-    pc: Pinecone,
+    qdrant: QdrantClient,
     session_id: str | None = None,
     top_k: int = 500,
     weights: dict | None = None,
@@ -38,21 +41,20 @@ async def stage1_retrieve(
     jd_text = f"{jd.get('title', '')} {jd.get('raw_text', '')}"
     query_vector = embed_query(jd_text)
 
-    pinecone_filter = _build_pinecone_filter(jd, session_id)
-    index = pc.Index(settings.index_name)
-    search_results = index.query(
-        vector=query_vector.tolist(),
-        filter=pinecone_filter,
-        top_k=top_k,
-        include_metadata=True,
+    qdrant_filter = _build_qdrant_filter(jd, session_id)
+    search_results = qdrant.search(
+        collection_name=settings.collection_name,
+        query_vector=query_vector.tolist(),
+        query_filter=qdrant_filter,
+        limit=top_k,
+        with_payload=True,
     )
 
-    if not search_results or not search_results.get("matches"):
+    if not search_results:
         return []
 
-    matches = search_results["matches"]
-    qdrant_ids = [r["id"] for r in matches]
-    score_by_qdrant_id = {r["id"]: float(r["score"]) for r in matches}
+    qdrant_ids = [r.id for r in search_results]
+    score_by_qdrant_id = {r.id: float(r.score) for r in search_results}
 
     result = await db.execute(select(Candidate).where(Candidate.qdrant_id.in_(qdrant_ids)))
     candidates = {c.qdrant_id: c for c in result.scalars().all()}

backend/src/routers/matching.py

@@ -23,8 +23,8 @@ from ..matching.scorer import rerank_with_weights
 router = APIRouter()
 
 
-def _get_pinecone(request: Request):
-    return request.app.state.pinecone
+def _get_qdrant(request: Request):
+    return request.app.state.qdrant
 
 
 async def _load_jd(jd_id: uuid.UUID, db: AsyncSession) -> JobDescription:
@@ -76,11 +76,11 @@ async def trigger_match(
     db: AsyncSession = Depends(get_db),
 ):
     jd = await _load_jd(jd_id, db)
-    pc = _get_pinecone(request)
+    qdrant = _get_qdrant(request)
     jd_dict = _build_jd_dict(jd)
     sid_str = str(session_id) if session_id else None
 
-    shortlist = await stage1_retrieve(jd_dict, db, pc, session_id=sid_str)
+    shortlist = await stage1_retrieve(jd_dict, db, qdrant, session_id=sid_str)
     final_ranked = await stage2_rerank(jd_dict, shortlist)
 
     await db.execute(
@@ -128,11 +128,11 @@ async def trigger_match(
 @router.get("/{jd_id}", response_model=MatchResponse)
 async def get_match_results(
     jd_id: uuid.UUID,
-    request: Request,
     session_id: uuid.UUID | None = Query(None),
     db: AsyncSession = Depends(get_db),
 ):
-    r = request.app.state.redis
+    settings = get_settings()
+    r = redis.Redis.from_url(settings.redis_url)
     cache_key = f"match_v2:{jd_id}:{session_id or 'none'}"
 
     try:

backend/src/routers/sessions.py

@@ -34,13 +34,14 @@ async def get_session(session_id: uuid.UUID, db: AsyncSession = Depends(get_db))
     return sess
 
 
-from pinecone import Pinecone
+from qdrant_client import QdrantClient
+from qdrant_client.models import Filter, FieldCondition, MatchValue
 from ..config import get_settings
 from ..models.match_result import MatchResult
 
-def _get_pinecone() -> Pinecone:
+def _get_qdrant() -> QdrantClient:
     settings = get_settings()
-    return Pinecone(api_key=settings.pinecone_api_key)
+    return QdrantClient(url=settings.qdrant_url, api_key=settings.qdrant_api_key)
 
 @router.delete("/{session_id}", status_code=204)
 async def delete_session(session_id: uuid.UUID, db: AsyncSession = Depends(get_db)):
@@ -53,11 +54,17 @@ async def delete_session(session_id: uuid.UUID, db: AsyncSession = Depends(get_d
     settings = get_settings()
     
     try:
-        pc = _get_pinecone()
-        index = pc.Index(settings.index_name)
-        index.delete(filter={"session_id": session_str})
+        qdrant = _get_qdrant()
+        qdrant.delete(
+            collection_name=settings.collection_name,
+            points_selector=Filter(
+                must=[
+                    FieldCondition(key="session_id", match=MatchValue(value=session_str))
+                ]
+            )
+        )
     except Exception as e:
-        print(f"Warning: Failed deleting Pinecone targets for session {session_str}: {e}")
+        print(f"Warning: Failed deleting Qdrant targets for session {session_str}: {e}")
 
     await db.execute(MatchResult.__table__.delete().where(MatchResult.session_id == session_id))
     await db.execute(Candidate.__table__.delete().where(Candidate.session_id == session_id))

backend/src/workers/celery_app.py

@@ -21,6 +21,4 @@ celery_app.conf.update(
     result_expires=3600,
     worker_prefetch_multiplier=1,
     task_acks_late=True,
-    broker_pool_limit=1,
-    redis_max_connections=5,
 )

backend/src/workers/ingest.py

@@ -4,7 +4,8 @@ import re
 import json
 from typing import Any
 from sqlalchemy import select, func
-from pinecone import Pinecone
+from qdrant_client import QdrantClient
+from qdrant_client.models import PointStruct
 
 from .celery_app import celery_app
 from ..config import get_settings
@@ -21,9 +22,9 @@ from ..ml.feature_builder import (
 )
 
 
-def _get_pinecone() -> Pinecone:
+def _get_qdrant() -> QdrantClient:
     settings = get_settings()
-    return Pinecone(api_key=settings.pinecone_api_key)
+    return QdrantClient(url=settings.qdrant_url, api_key=settings.qdrant_api_key)
 
 
 def _normalize_bool(val: Any) -> bool | None:
@@ -73,8 +74,7 @@ def ingest_candidates_batch(self, rows: list[dict], session_id: str | None = Non
 
 async def _ingest_candidates_async(rows: list[dict], session_id: str | None) -> dict:
     settings = get_settings()
-    pc = _get_pinecone()
-    index = pc.Index(settings.index_name)
+    qdrant = _get_qdrant()
     texts = []
     processed = []
     sess_uuid = uuid.UUID(session_id) if session_id else None
@@ -101,7 +101,7 @@ async def _ingest_candidates_async(rows: list[dict], session_id: str | None) ->
     embeddings = embed_texts(texts)
 
     async with AsyncSessionLocal() as session:
-        pinecone_vectors = []
+        qdrant_points = []
         inserted = 0
 
         for i, p in enumerate(processed):
@@ -154,30 +154,32 @@ async def _ingest_candidates_async(rows: list[dict], session_id: str | None) ->
             )
             session.add(cand)
 
-            pinecone_vectors.append({
-                "id": qdrant_id,
-                "values": embeddings[i].tolist(),
-                "metadata": {
-                    "candidate_id": str(candidate_id),
-                    "session_id": session_id or "",
-                    "role_type": row.get("role_type") or "",
-                    "engineer_type": row.get("engineer_type") or "",
-                    "years_of_experience": _normalize_float(row.get("years_of_experience")) or 0.0,
-                    "looking_for": row.get("looking_for") or "",
-                    "open_to_working_at": row.get("open_to_working_at") or "",
-                    "skills": [str(s) for s in all_skills[:50]],
-                    "gen_ai_experience": bool(_normalize_bool(row.get("gen_ai_experience"))),
-                    "growth_velocity": float(p["velocity"]) if p["velocity"] is not None else 0.0,
-                    "is_funded": bool(_normalize_bool(row.get("most_recent_company_is_funded"))),
-                    "currently_employed": bool(_normalize_bool(row.get("currently_employed"))),
-                },
-            })
+            qdrant_points.append(
+                PointStruct(
+                    id=qdrant_id,
+                    vector=embeddings[i].tolist(),
+                    payload={
+                        "candidate_id": str(candidate_id),
+                        "session_id": session_id or "",
+                        "role_type": row.get("role_type"),
+                        "engineer_type": row.get("engineer_type"),
+                        "years_of_experience": _normalize_float(row.get("years_of_experience")),
+                        "looking_for": row.get("looking_for"),
+                        "open_to_working_at": row.get("open_to_working_at"),
+                        "skills": all_skills[:50],
+                        "gen_ai_experience": _normalize_bool(row.get("gen_ai_experience")),
+                        "growth_velocity": p["velocity"],
+                        "is_funded": _normalize_bool(row.get("most_recent_company_is_funded")),
+                        "currently_employed": _normalize_bool(row.get("currently_employed")),
+                    },
+                )
+            )
             inserted += 1
 
         await session.commit()
 
-        if pinecone_vectors:
-            index.upsert(vectors=pinecone_vectors)
+        if qdrant_points:
+            qdrant.upsert(collection_name=settings.collection_name, points=qdrant_points)
 
         if sess_uuid:
             result = await session.execute(

backend/test_pinecone.py

@@ -1,36 +0,0 @@
-import os
-from pinecone import Pinecone
-from dotenv import load_dotenv
-
-# Load .env file
-load_dotenv()
-
-api_key = os.getenv("PINECONE_API_KEY")
-
-if not api_key:
-    print("Error: PINECONE_API_KEY not found in .env file")
-    exit(1)
-
-print(f"Connecting to Pinecone with key: {api_key[:10]}...")
-
-try:
-    pc = Pinecone(api_key=api_key)
-    
-    # Testing general connectivity by listing indexes
-    print("Listing indexes...")
-    indexes = pc.list_indexes()
-    print(f"Successfully connected! Found {len(indexes)} indexes.")
-    for idx in indexes:
-        print(f" - {idx.name}")
-        
-    # Testing Assistant API connectivity
-    print("\nTesting Assistant API...")
-    assistants = pc.assistant.list_assistants()
-    print(f"Successfully reached Assistant API! Found {len(assistants)} assistants.")
-    for a in assistants:
-        print(f" - {a.name} ({a.status})")
-
-    print("\nConnection test passed successfully!")
-
-except Exception as e:
-    print(f"\nConnection failed: {e}")