Phase 10: Capstone Deployment Project · 8-12 hours · Python · LangGraph · Pinecone
The Problem
The capstone project is the portfolio artifact that gets FDE interviews. You'll build a complete enterprise RAG + agent system: document processing, hybrid search, access control, LangGraph agent with enterprise tool integration, Kubernetes deployment, evaluation harness with RAGAS metrics, and a client presentation. This lesson is the project kickoff.
Capstone project architecture
from dataclasses import dataclass, field
class CapstoneArchitecture:
"""Complete architecture for the capstone enterprise RAG + agent system.
This is the system you'll build, deploy, evaluate, and present.
"""
COMPONENTS = {
"document_processing": {
"ingestion_sources": ["SharePoint", "Confluence", "PDF files", "CRM notes"],
"pipeline": [
"Load documents from source connectors",
"Extract text (PyMuPDF for PDFs, BeautifulSoup for HTML)",
"Chunk with recursive splitter (chunk_size=512, overlap=64)",
"Generate embeddings (text-embedding-3-large, 3072 dims)",
"Store in pgvector with metadata (source, access_level, timestamp)",
],
"access_control": "Metadata-based filtering: each chunk tagged with access_level (public, internal, confidential, restricted)",
},
"retrieval": {
"strategy": "Hybrid search: BM25 keyword + pgvector cosine similarity",
"reranking": "Cohere reranker API (top 20 → top 5)",
"query_transform": "HyDE (Hypothetical Document Embeddings) for better semantic match",
"fallback": "If vector search returns <3 results, fall back to BM25-only",
},
"generation": {
"primary_model": "GPT-4o for complex queries (detected by query classifier)",
"secondary_model": "GPT-4o-mini for simple queries (70% of traffic)",
"fallback_chain": ["GPT-4o", "Claude 3.5 Sonnet", "Cached response"],
"guardrails": [
"PII detection (regex + NER model)",
"Topic restriction (only answer from retrieved context)",
"Response length limit (500 tokens max)",
],
},
"agent_layer": {
"framework": "LangGraph",
"tools": [
{"name": "search_knowledge_base", "description": "Search internal documents"},
{"name": "lookup_crm_record", "description": "Query Salesforce CRM by customer ID"},
{"name": "create_ticket", "description": "Create ServiceNow ticket"},
{"name": "escalate_to_human", "description": "Route to human agent with context"},
],
"orchestration": "ReAct pattern: reason → act → observe → repeat until done",
"max_iterations": 5,
"timeout": 30,
},
"deployment": {
"container": "Docker (Python 3.12-slim base)",
"orchestration": "Kubernetes (Deployment + HPA + Service)",
"replicas": "2-10 (auto-scaled on CPU > 70%)",
"ingress": "nginx-ingress with TLS termination",
"config": "Kubernetes Secrets for API keys, ConfigMap for non-sensitive config",
},
"observability": {
"tracing": "LangSmith for LLM call traces and retrieval quality",
"metrics": "Prometheus + Grafana (latency, error rate, cache hit rate)",
"logging": "Structured JSON logs to stdout (collected by Fluentd)",
"alerting": "PagerDuty for SEV1, Slack for SEV2-3",
},
}
Building the RAG pipeline
class CapstoneRAGPipeline:
"""Core RAG pipeline for the capstone project."""
def __init__(self, db_url: str, embedding_model: str = "text-embedding-3-large"):
self.db_url = db_url
self.embedding_model = embedding_model
def ingest_documents(self, documents: list[dict]) -> dict:
"""Ingest documents: chunk, embed, store with metadata.
Args:
documents: list of {"content": str, "source": str, "access_level": str}
Returns:
ingestion report with chunk count and timing
"""
from langchain_text_splitters import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=512,
chunk_overlap=64,
separators=["\n\n", "\n", ". ", " ", ""],
)
total_chunks = 0
for doc in documents:
chunks = splitter.split_text(doc["content"])
for i, chunk in enumerate(chunks):
# In production: embed and insert into pgvector
# embedding = openai.embeddings.create(
# model=self.embedding_model, input=chunk
# ).data[0].embedding
# cursor.execute(
# "INSERT INTO documents (content, embedding, source, access_level, chunk_idx) "
# "VALUES (%s, %s, %s, %s, %s)",
# (chunk, embedding, doc["source"], doc["access_level"], i)
# )
total_chunks += 1
return {
"documents_ingested": len(documents),
"chunks_created": total_chunks,
"avg_chunks_per_doc": total_chunks / len(documents) if documents else 0,
}
def retrieve(self, query: str, user_access_level: str, top_k: int = 5) -> list[dict]:
"""Hybrid retrieval: BM25 + vector search with access control filtering.
Args:
query: user query
user_access_level: "public", "internal", "confidential", or "restricted"
top_k: number of chunks to return
Returns:
list of retrieved chunks with relevance scores
"""
# Step 1: Generate query embedding
# query_embedding = openai.embeddings.create(
# model=self.embedding_model, input=query
# ).data[0].embedding
# Step 2: Vector search with access control filter
# vector_results = cursor.execute(
# """SELECT content, source, access_level,
# embedding <=> %s AS distance
# FROM documents
# WHERE access_level <= %s
# ORDER BY distance LIMIT %s""",
# (query_embedding, user_access_level, top_k * 2)
# ).fetchall()
# Step 3: BM25 keyword search with same access filter
# bm25_results = cursor.execute(
# """SELECT content, source, access_level,
# ts_rank_cd(tsv, plainto_tsquery(%s)) AS rank
# FROM documents
# WHERE access_level <= %s AND tsv @@ plainto_tsquery(%s)
# ORDER BY rank DESC LIMIT %s""",
# (query, user_access_level, query, top_k * 2)
# ).fetchall()
# Step 4: Merge and rerank
# merged = self._merge_scores(vector_results, bm25_results)
# reranked = self._rerank(query, merged, top_k)
# return reranked
pass
def generate(self, query: str, retrieved_chunks: list[dict]) -> dict:
"""Generate response with guardrails.
Returns:
{"answer": str, "sources": list, "model": str, "tokens": int}
"""
# Build prompt with retrieved context
context = "\n\n".join(
f"[Source: {c['source']}]\n{c['content']}"
for c in retrieved_chunks
)
system_prompt = f"""You are an enterprise assistant. Answer ONLY based on the provided context.
If the context doesn't contain the answer, say "I don't have enough information to answer this question."
Context:
{context}"""
# Choose model based on query complexity
is_complex = len(query) > 100 or any(
word in query.lower()
for word in ["analyze", "compare", "summarize", "explain", "why"]
)
model = "gpt-4o" if is_complex else "gpt-4o-mini"
# Call with fallback chain
# response = self._call_with_fallback(
# model=model,
# messages=[
# {"role": "system", "content": system_prompt},
# {"role": "user", "content": query},
# ],
# max_tokens=500,
# )
# Guardrail check: PII detection on response
# if self._contains_pii(response.content):
# return {"answer": "I cannot provide that information.", "blocked": True}
return {
"answer": "[generated response]",
"sources": [c["source"] for c in retrieved_chunks],
"model": model,
"tokens": 0,
}
LangGraph agent integration
class CapstoneAgent:
"""LangGraph agent with enterprise tool integration."""
TOOLS = [
{
"name": "search_knowledge_base",
"description": "Search internal documents for answers",
"parameters": {"query": "str", "access_level": "str"},
},
{
"name": "lookup_crm_record",
"description": "Look up customer information in Salesforce",
"parameters": {"customer_id": "str"},
},
{
"name": "create_ticket",
"description": "Create a ServiceNow ticket for issue tracking",
"parameters": {"title": "str", "description": "str", "priority": "str"},
},
{
"name": "escalate_to_human",
"description": "Escalate complex issue to human agent",
"parameters": {"reason": "str", "context": "str"},
},
]
def build_graph(self):
"""Build the LangGraph state machine.
States: START → reason → act → observe → (reason | END)
"""
# from langgraph.graph import StateGraph, END
#
# graph = StateGraph(AgentState)
# graph.add_node("reason", self._reason_node)
# graph.add_node("act", self._act_node)
# graph.add_node("observe", self._observe_node)
#
# graph.set_entry_point("reason")
# graph.add_edge("reason", "act")
# graph.add_edge("act", "observe")
# graph.add_conditional_edges(
# "observe",
# self._should_continue,
# {"continue": "reason", "stop": END},
# )
#
# return graph.compile()
pass
def _should_continue(self, state) -> str:
"""Decide whether to continue iterating or return final answer."""
if state["iteration_count"] >= 5:
return "stop"
if state.get("final_answer"):
return "stop"
return "continue"
Kubernetes deployment
# k8s/deployment.yaml — Capstone system deployment
apiVersion: apps/v1
kind: Deployment
metadata:
name: capstone-rag
spec:
replicas: 2
selector:
matchLabels:
app: capstone-rag
template:
metadata:
labels:
app: capstone-rag
spec:
containers:
- name: api
image: capstone-rag:latest
ports:
- containerPort: 8000
env:
- name: OPENAI_API_KEY
valueFrom:
secretKeyRef:
name: api-secrets
key: openai-key
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: db-secrets
key: url
resources:
requests:
cpu: 500m
memory: 1Gi
limits:
cpu: 1000m
memory: 2Gi
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 10
periodSeconds: 5
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 30
periodSeconds: 10
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: capstone-rag-hpa
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: capstone-rag
minReplicas: 2
maxReplicas: 10
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70