Advanced: RAG, AI SaaS Backend, Real-time Recommender

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct, VectorParams, Distance
import uuid, pypdf

qdrant = QdrantClient(url="http://qdrant:6333")
emb = OpenAIEmbeddings(model="text-embedding-3-small")

def ingest(user_id: str, pdf_path: str):
    text = "".join(p.extract_text() for p in pypdf.PdfReader(pdf_path).pages)
    chunks = RecursiveCharacterTextSplitter(
        chunk_size=800, chunk_overlap=100
    ).split_text(text)
    vectors = emb.embed_documents(chunks)

    collection = f"docs_{user_id}"
    if not qdrant.collection_exists(collection):
        qdrant.create_collection(collection,
            vectors_config=VectorParams(size=1536, distance=Distance.COSINE))

    qdrant.upsert(collection, points=[
        PointStruct(id=str(uuid.uuid4()), vector=v,
                    payload={"text": c, "source": pdf_path})
        for c, v in zip(chunks, vectors)
    ])

@app.post("/ask")
def ask(req: AskReq):
    qvec = emb.embed_query(req.question)
    hits = qdrant.search(f"docs_{req.user_id}", query_vector=qvec, limit=4)
    context = "\n---\n".join(f"[{i+1}] {h.payload['text']}" for i, h in enumerate(hits))

    prompt = f"""Answer based ONLY on context. Cite [n].
Context:
{context}

Q: {req.question}
A:"""
    answer = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role":"user","content":prompt}],
    ).choices[0].message.content
    return {"answer": answer, "sources": [h.payload["source"] for h in hits]}

# usage_tracker.py
from datetime import datetime
import redis
r = redis.Redis()

def track(user_id: str, tokens_in: int, tokens_out: int):
    month = datetime.utcnow().strftime("%Y-%m")
    pipe = r.pipeline()
    pipe.hincrby(f"usage:{user_id}:{month}", "tokens_in", tokens_in)
    pipe.hincrby(f"usage:{user_id}:{month}", "tokens_out", tokens_out)
    pipe.hincrby(f"usage:{user_id}:{month}", "calls", 1)
    pipe.execute()

def check_quota(user_id: str, plan: str) -> bool:
    month = datetime.utcnow().strftime("%Y-%m")
    used = int(r.hget(f"usage:{user_id}:{month}", "tokens_in") or 0)
    return used < PLAN_LIMITS[plan]

# Stripe usage record
stripe.SubscriptionItem.create_usage_record(
    sub_item_id, quantity=tokens_in + tokens_out,
    timestamp=int(time.time()), action="increment",
)

@app.get("/recommend/{user_id}")
async def recommend(user_id: str):
    # 1. Online features from Redis (last 5 click vec avg)
    uvec = await redis.get(f"uvec:{user_id}") or default_vec

    # 2. ANN candidate (top 200)
    candidates = faiss_index.search(uvec.reshape(1, -1), k=200)[1][0]

    # 3. Feature fetch + rank
    feats = feature_store.batch_get(user_id, candidates)
    scores = ranker.predict(feats)

    # 4. Diversify (MMR) + top 20
    top = mmr_rerank(candidates, scores, k=20, lambda_=0.7)

    # 5. Log impression for training
    kafka.send("impressions", {"user": user_id, "items": top, "ts": time.time()})
    return {"items": top}