Capstone Project
The capstone project brings together all GenAI concepts into a production-ready system with RAG, agents, guardrails, evaluation, and monitoring.
Project Architecture
from dataclasses import dataclass
from typing import Dict, List, Optional
import json
@dataclass
class ProjectConfig:
name: str
description: str
data_sources: List[str]
model_config: Dict
rag_config: Dict
guardrails_config: Dict
monitoring_config: Dict
class GenAIProject:
def __init__(self, config: ProjectConfig):
self.config = config
self.components = {}
self.initialized = False
def initialize(self):
self.components["ingestion"] = self._setup_ingestion()
self.components["vector_store"] = self._setup_vector_store()
self.components["retriever"] = self._setup_retriever()
self.components["llm"] = self._setup_llm()
self.components["agent"] = self._setup_agent()
self.components["guardrails"] = self._setup_guardrails()
self.components["evaluator"] = self._setup_evaluator()
self.components["monitor"] = self._setup_monitor()
self.initialized = True
def _setup_ingestion(self):
from langchain.text_splitter import RecursiveCharacterTextSplitter
return {
"splitter": RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50),
"embeddings": None # Set up embeddings
}
def _setup_vector_store(self):
return {"type": "faiss", "index": None}
def _setup_retriever(self):
return {"top_k": 5, "reranker": None}
def _setup_llm(self):
from langchain_openai import ChatOpenAI
return ChatOpenAI(model="gpt-4", temperature=0)
def _setup_agent(self):
return {"tools": [], "max_iterations": 5}
def _setup_guardrails(self):
return {"input_filter": None, "output_filter": None}
def _setup_evaluator(self):
return {"metrics": ["faithfulness", "relevancy"]}
def _setup_monitor(self):
return {"log_path": "logs/", "alert_threshold": 0.8}
def process_query(self, query: str) -> dict:
if not self.initialized:
raise RuntimeError("Project not initialized")
monitored_query = self.components["guardrails"].get("input_filter")
if monitored_query:
query = monitored_query(query)
retrieval_result = self.components["retriever"]
context = self._retrieve_context(query)
response = self.components["llm"].invoke(
f"Context: {context}\nQuestion: {query}\nAnswer:"
).content
monitored_response = self.components["guardrails"].get("output_filter")
if monitored_response:
response = monitored_response(response)
self._log_interaction(query, response)
return {"answer": response, "context": context}
def _retrieve_context(self, query: str) -> str:
return "Retrieved context from vector store"
def _log_interaction(self, query: str, response: str):
print(f"Query: {query[:50]}... Response: {response[:50]}...")
# Usage
config = ProjectConfig(
name="Customer Support AI",
description="AI-powered customer support system",
data_sources=["knowledge_base.pdf", "faq.json"],
model_config={"model": "gpt-4", "temperature": 0},
rag_config={"top_k": 5, "chunk_size": 500},
guardrails_config={"toxicity_threshold": 0.8},
monitoring_config={"log_all": True}
)
project = GenAIProject(config)
project.initialize()
Data Pipeline
class DataPipeline:
def __init__(self, project: GenAIProject):
self.project = project
def ingest_documents(self, file_paths: list):
from langchain_community.document_loaders import (
PyPDFLoader, TextLoader, CSVLoader
)
documents = []
for path in file_paths:
if path.endswith(".pdf"):
loader = PyPDFLoader(path)
elif path.endswith(".txt"):
loader = TextLoader(path)
elif path.endswith(".csv"):
loader = CSVLoader(path)
documents.extend(loader.load())
splitter = self.project.components["ingestion"]["splitter"]
chunks = splitter.split_documents(documents)
return chunks
def create_embeddings(self, chunks: list):
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
embeddings = OpenAIEmbeddings()
vector_store = FAISS.from_documents(chunks, embeddings)
self.project.components["vector_store"]["index"] = vector_store
return vector_store
def run_pipeline(self, file_paths: list):
chunks = self.ingest_documents(file_paths)
vector_store = self.create_embeddings(chunks)
return {"chunks": len(chunks), "status": "success"}
# Usage
pipeline = DataPipeline(project)
result = pipeline.run_pipeline(["docs/knowledge.pdf", "docs/faq.txt"])
Evaluation Suite
class EvaluationSuite:
def __init__(self, project: GenAIProject):
self.project = project
self.test_cases = []
def add_test_case(self, question: str, expected_answer: str,
context: str = None):
self.test_cases.append({
"question": question,
"expected": expected_answer,
"context": context
})
def run_evaluation(self) -> dict:
results = []
for case in self.test_cases:
response = self.project.process_query(case["question"])
score = self._score_response(response["answer"], case["expected"])
results.append({
"question": case["question"],
"expected": case["expected"],
"actual": response["answer"],
"score": score
})
avg_score = sum(r["score"] for r in results) / len(results)
return {"results": results, "average_score": avg_score}
def _score_response(self, actual: str, expected: str) -> float:
actual_words = set(actual.lower().split())
expected_words = set(expected.lower().split())
if not expected_words:
return 0.0
overlap = actual_words & expected_words
return len(overlap) / len(expected_words)
def generate_report(self) -> str:
eval_results = self.run_evaluation()
report = f"Evaluation Report\n{'='*40}\n"
report += f"Total Test Cases: {len(self.test_cases)}\n"
report += f"Average Score: {eval_results['average_score']:.2f}\n\n"
for result in eval_results["results"]:
status = "PASS" if result["score"] > 0.7 else "FAIL"
report += f"[{status}] {result['question'][:50]}...\n"
return report
# Usage
eval_suite = EvaluationSuite(project)
eval_suite.add_test_case("What is your return policy?", "30 days return")
report = eval_suite.generate_report()
Deployment Configuration
class DeploymentConfig:
def __init__(self):
self.environment = "production"
self.scaling = {"min_replicas": 2, "max_replicas": 10}
self.resources = {"cpu": "2", "memory": "8Gi", "gpu": "1"}
def to_kubernetes(self) -> dict:
return {
"apiVersion": "apps/v1",
"kind": "Deployment",
"metadata": {"name": "genai-app"},
"spec": {
"replicas": self.scaling["min_replicas"],
"selector": {"matchLabels": {"app": "genai"}},
"template": {
"spec": {
"containers": [{
"name": "genai",
"image": "genai-app:latest",
"resources": {
"requests": self.resources,
"limits": self.resources
}
}]
}
}
}
}
def to_docker_compose(self) -> str:
return """
version: '3.8'
services:
genai-app:
build: .
ports:
- "8000:8000"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
deploy:
replicas: 2
"""
# Usage
deploy = DeploymentConfig()
k8s_config = deploy.to_kubernetes()
docker_config = deploy.to_docker_compose()
Key Takeaways
- Architecture integrates all components into a cohesive system
- Data pipelines handle ingestion, chunking, and embedding
- RAG + Agents provide intelligent retrieval and reasoning
- Guardrails ensure safety and compliance
- Evaluation validates system quality with metrics
- Deployment scales the system for production use
- Monitoring tracks performance and costs in real-time