Conversational AI Agent
for UPSC Aspirants

@app.post("/chat")
async def chat(req: ChatRequest, user: User = Depends(verify_jwt)):
    # Rate limit: 100 queries/student/day — atomic Redis INCR
    rate_key = f"rate:{user.id}:{today()}"
    if await redis.incr(rate_key) > 100:
        raise HTTPException(429, "Daily query limit reached")
    await redis.expire(rate_key, 86400)

    # Semantic cache lookup — zero GPT-4 cost on hit
    cached = await semantic_cache.lookup(req.query)
    if cached:
        return StreamingResponse(stream_string(cached))

    # Dispatch to Celery — return task ID in <5ms, never block
    task = rag_pipeline.delay(req.query, req.session_id, user.id)
    return {"task_id": task.id}

@app.get("/stream/{task_id}")
async def stream_result(task_id: str):
    # SSE stream: tokens arrive as worker produces them
    return StreamingResponse(
        token_generator(task_id),
        media_type="text/event-stream",
    )

# Job 1: Semantic cache — query Pinecone cache namespace
async def cache_lookup(query: str) -> str | None:
    embedding = await embed(query)
    results = pinecone_index.query(
        vector=embedding, top_k=1, namespace="cache",
        include_metadata=True,
    )
    if results.matches and results.matches[0].score >= 0.92:
        return results.matches[0].metadata["answer"]
    return None

# Job 2: Rate limiter — atomic, no race conditions
async def check_rate_limit(user_id: str) -> bool:
    key = f"rate:{user_id}:{today()}"
    count = await redis.incr(key)
    if count == 1:
        await redis.expire(key, 86400)  # reset daily
    return count <= 100

# Job 3: Session memory — bounded 10-turn window, 30-min TTL
async def get_history(session_id: str) -> list[dict]:
    raw = await redis.lrange(f"session:{session_id}", 0, 9)
    return [json.loads(m) for m in reversed(raw)]

async def append_turn(session_id: str, message: dict):
    key = f"session:{session_id}"
    await redis.lpush(key, json.dumps(message))
    await redis.ltrim(key, 0, 9)     # enforce 10-turn limit
    await redis.expire(key, 1800)    # 30-min inactivity TTL

@celery.task(bind=True, max_retries=3)
def rag_pipeline(self, query: str, session_id: str, user_id: str):
    try:
        # Restore conversation context from Redis
        history = get_session_history_sync(session_id)

        # Invoke LangGraph supervisor agent
        result = supervisor_graph.invoke({
            "query": query,
            "history": history,
            "user_id": user_id,
        })

        # Cache answer in Pinecone cache namespace
        pinecone_index.upsert(
            vectors=[{
                "id": cache_id(query),
                "values": embed_sync(query),
                "metadata": {"answer": result["answer"]},
            }],
            namespace="cache",
        )

        # Append turn to session memory
        append_turn_sync(session_id, {
            "role": "assistant", "content": result["answer"]
        })
        return result

    except openai.RateLimitError as e:
        raise self.retry(exc=e, countdown=5)
    except Exception as e:
        raise self.retry(exc=e, countdown=2)

from langgraph.graph import StateGraph, END
from typing import TypedDict

class AgentState(TypedDict):
    query:            str
    history:          list[dict]
    intent:           str           # set by supervisor
    retrieved_chunks: list[str]
    answer:           str
    sources:          list[str]
    guardrail_passed: bool
    token_usage:      dict          # per-node cost attribution

def supervisor_node(state: AgentState) -> AgentState:
    intent = classify_intent(state["query"], state["history"])
    return {**state, "intent": intent}

def route(state: AgentState) -> str:
    return state["intent"]  # "factual" | "pyq" | "essay" | "current_affairs"

def factual_agent(state: AgentState) -> AgentState:
    chunks = pinecone_query(state["query"], namespace="factual", top_k=5)
    answer = gpt4_generate(FACTUAL_PROMPT, chunks, state["history"])
    return {**state, "retrieved_chunks": chunks, "answer": answer}

graph = StateGraph(AgentState)
graph.add_node("supervisor",      supervisor_node)
graph.add_node("factual",         factual_agent)
graph.add_node("pyq",             pyq_agent)
graph.add_node("essay",           essay_agent)
graph.add_node("current_affairs", current_affairs_agent)
graph.set_entry_point("supervisor")
graph.add_conditional_edges("supervisor", route, {
    "factual": "factual", "pyq": "pyq",
    "essay": "essay", "current_affairs": "current_affairs",
})
supervisor_graph = graph.compile()

# Namespace-per-domain: each subject isolated
DOMAIN_NAMESPACES = {
    "factual": ["polity", "history", "economy", "geography"],
    "pyq":     ["pyq_prelims", "pyq_mains"],
    "essay":   ["essay_structure", "essay_content"],
    "current": ["current_affairs"],
    "cache":   ["cache"],
}

def upsert_document(doc: Document, domain: str):
    """Hierarchical: child chunks indexed, parent assembled for context."""
    for i, parent in enumerate(chunk(doc.text, size=512, overlap=50)):
        for j, child in enumerate(chunk(parent, size=128, overlap=20)):
            pinecone_index.upsert(vectors=[{
                "id": f"{doc.id}_p{i}_c{j}",
                "values": embed(child),
                "metadata": {
                    "child_text":  child,
                    "parent_text": parent,   # full context for GPT-4
                    "domain":      domain,
                    "updated_at":  doc.updated_at,
                },
            }], namespace=domain)

def retrieve(query: str, namespace: str, top_k: int = 5) -> list[str]:
    """Returns parent text — complete context — for matched child chunks."""
    results = pinecone_index.query(
        vector=embed(query), top_k=top_k,
        namespace=namespace, include_metadata=True,
    )
    return [m.metadata["parent_text"] for m in results.matches]

# Dockerfile.api — multi-stage, lean gateway image (~190MB)
FROM python:3.11-slim AS builder
WORKDIR /app
COPY requirements/api.txt .
RUN pip install --no-cache-dir -r api.txt

FROM python:3.11-slim
WORKDIR /app
COPY --from=builder /usr/local/lib/python3.11 /usr/local/lib/python3.11
COPY src/ .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# ── docker-compose.yml (local = production parity) ────────────
services:
  api:
    build: { context: ., dockerfile: Dockerfile.api }
    env_file: .env.local
    ports: ["8000:8000"]
    depends_on: [redis]

  worker:
    build: { context: ., dockerfile: Dockerfile.worker }
    command: celery -A tasks worker --concurrency=8 --loglevel=info
    env_file: .env.local
    depends_on: [redis]

  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]

# .github/workflows/deploy.yml
jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run RAGAS eval suite  # blocks deploy on threshold failure
        run: |
          python evals/run_ragas.py
            --dataset evals/golden_set.json
            --thresholds '{"faithfulness":0.90,"answer_relevancy":0.88,"accuracy":0.93}'

  test:
    needs: evaluate
    steps:
      - run: pytest tests/ -v --cov=app --cov-fail-under=80

  build-push:
    needs: test
    steps:
      - name: Build and push API + worker images to ECR
        run: |
          docker build -f Dockerfile.api    -t $ECR_REGISTRY/upsc-api:$GITHUB_SHA .
          docker build -f Dockerfile.worker -t $ECR_REGISTRY/upsc-worker:$GITHUB_SHA .
          docker push $ECR_REGISTRY/upsc-api:$GITHUB_SHA
          docker push $ECR_REGISTRY/upsc-worker:$GITHUB_SHA

  deploy:
    needs: build-push
    steps:
      - name: Blue-green ECS deploy (old version stays live until health checks pass)
        run: |
          aws ecs update-service
            --cluster upsc-prod --service upsc-api
            --task-definition upsc-api:$NEW_REVISION
            --deployment-configuration minimumHealthyPercent=100,maximumPercent=200

from langsmith import Client, traceable
from langsmith.run_helpers import get_current_run_tree

ls = Client()

# Pin prompt versions — never "latest" in production
SUPERVISOR_PROMPT = ls.pull_prompt("upsc-supervisor:prod-v1.4")
FACTUAL_PROMPT    = ls.pull_prompt("upsc-factual:prod-v2.1")

@traceable(name="factual_agent", tags=["specialist", "factual"])
def factual_agent(state: AgentState) -> AgentState:
    run = get_current_run_tree()

    chunks = retrieve(state["query"], namespace="factual")
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=build_messages(FACTUAL_PROMPT, chunks, state["history"]),
    )

    # Token usage tracked per node for cost attribution
    run.extra["token_usage"] = response.usage.model_dump()
    return {**state, "answer": response.choices[0].message.content}

# User feedback captured per run for prompt refinement
def record_feedback(run_id: str, score: int, comment: str = ""):
    ls.create_feedback(
        run_id=run_id,
        key="user_rating",
        score=score,          # 1 = thumbs up, 0 = thumbs down
        comment=comment,
    )