It is working
# ============================================================
# STEP 1 — INSTALL REQUIRED LIBRARIES
# ============================================================
# Run this in Google Colab
!pip install -q google-generativeai numpy
# ============================================================
# STEP 2 — IMPORT LIBRARIES
# ============================================================
import os
import numpy as np
import google.generativeai as genai
from google.colab import userdata
# ============================================================
# STEP 3 — CONFIGURE GEMINI API
# ============================================================
# Store your API key in:
# Colab Secrets → GEMINI_API_KEY
GEMINI_API_KEY = userdata.get("GEMINI_API_KEY")
genai.configure(api_key=GEMINI_API_KEY)
# ============================================================
# STEP 4 — CREATE DATASET
# ============================================================
documents = [
# --------------------------------------------------------
# POD FAILURES / DEBUGGING
# --------------------------------------------------------
"CrashLoopBackOff occurs when a container repeatedly crashes after starting.",
"OOMKilled happens when a container exceeds its memory limit.",
"A container may crash due to missing environment variables.",
"Incorrect command or entrypoint can cause container startup failure.",
"Application errors inside the container often lead to restarts.",
"kubectl logs retrieves logs from a running container.",
"kubectl describe pod shows events and state transitions.",
"Liveness probes determine if a container should be restarted.",
"Readiness probes determine if a pod can receive traffic.",
# --------------------------------------------------------
# SCHEDULING
# --------------------------------------------------------
"Pods remain pending if no node satisfies resource requests.",
"Node affinity restricts pods to specific nodes.",
"Taints prevent pods from being scheduled on certain nodes.",
"Tolerations allow pods to be scheduled on tainted nodes.",
# --------------------------------------------------------
# SERVICES
# --------------------------------------------------------
"ClusterIP services expose applications within the cluster.",
"NodePort services expose applications on node IPs.",
"LoadBalancer services expose applications externally.",
"Ingress routes HTTP and HTTPS traffic to services.",
# --------------------------------------------------------
# STORAGE
# --------------------------------------------------------
"PersistentVolumes provide storage independent of pods.",
"PersistentVolumeClaims request storage resources.",
"StorageClasses define dynamic provisioning behavior.",
# --------------------------------------------------------
# DEPLOYMENTS
# --------------------------------------------------------
"Deployments manage replica sets and pod updates.",
"Rolling updates gradually replace old pods with new ones.",
"ReplicaSets maintain a stable number of pod replicas.",
# --------------------------------------------------------
# CONFIGURATION
# --------------------------------------------------------
"ConfigMaps store non-sensitive configuration data.",
"Secrets store sensitive data like passwords and tokens.",
"Environment variables can be injected from ConfigMaps and Secrets.",
# --------------------------------------------------------
# IMAGES / REGISTRY
# --------------------------------------------------------
"ImagePullBackOff occurs when Kubernetes cannot pull the container image.",
"Incorrect image name or tag can cause image pull failures.",
"Private registries require imagePullSecrets for authentication.",
# --------------------------------------------------------
# AUTOSCALING
# --------------------------------------------------------
"Horizontal Pod Autoscaler scales based on CPU or metrics.",
# --------------------------------------------------------
# SECURITY
# --------------------------------------------------------
"RBAC controls access permissions inside Kubernetes.",
"RBAC misconfiguration can block access to resources.",
# --------------------------------------------------------
# NETWORKING
# --------------------------------------------------------
"NetworkPolicies control communication between pods.",
# --------------------------------------------------------
# CLEANUP
# --------------------------------------------------------
"Pods stuck in Terminating state may have finalizers blocking deletion."
]
print(f"Total documents: {len(documents)}")
# ============================================================
# STEP 5 — CREATE SLIDING WINDOW CHUNKS
# ============================================================
# WHY?
# ----
# Instead of KMeans clustering, we preserve
# natural neighboring context.
#
# Example:
# sentence1 + sentence2
# sentence2 + sentence3
#
# This is:
# - automatic
# - scalable
# - semantically safer
WINDOW_SIZE = 3
STRIDE = 1
smart_chunks = []
for i in range(0, len(documents) - WINDOW_SIZE + 1, STRIDE):
chunk = documents[i:i + WINDOW_SIZE]
chunk_text = "\n".join(chunk)
smart_chunks.append(chunk_text)
print(f"Total chunks created: {len(smart_chunks)}")
# ============================================================
# STEP 6 — PREPARE STRUCTURED CHUNK DATA
# ============================================================
prepared_data = []
for i, chunk in enumerate(smart_chunks):
prepared_data.append({
"id": f"chunk_{i}",
"text": chunk
})
print(f"Prepared chunks: {len(prepared_data)}")
# ============================================================
# STEP 7 — CREATE EMBEDDING FUNCTION
# ============================================================
def get_embedding(text):
response = genai.embed_content(
model="models/gemini-embedding-001",
content=text
)
return response["embedding"]
# ============================================================
# STEP 8 — GENERATE CHUNK EMBEDDINGS
# ============================================================
print("Generating embeddings...")
for item in prepared_data:
embedding = get_embedding(item["text"])
item["embedding"] = embedding
print("Embeddings added successfully.")
# ============================================================
# STEP 9 — NORMALIZATION FUNCTION
# ============================================================
# WHY?
# ----
# Makes cosine similarity more stable
def normalize(vec):
vec = np.array(vec)
return vec / np.linalg.norm(vec)
# ============================================================
# STEP 10 — COSINE SIMILARITY FUNCTION
# ============================================================
def cosine_similarity(a, b):
a = normalize(a)
b = normalize(b)
return np.dot(a, b)
# ============================================================
# STEP 11 — RETRIEVAL FUNCTION
# ============================================================
# FEATURES:
# ----------
# 1. Query embedding
# 2. Cosine similarity
# 3. Threshold filtering
# 4. Re-ranking using keyword bonus
def retrieve(query, top_k=3, min_score=0.55):
# --------------------------------------------------------
# EMBED QUERY
# --------------------------------------------------------
query_embedding = get_embedding(query)
scores = []
# --------------------------------------------------------
# COMPARE AGAINST ALL CHUNKS
# --------------------------------------------------------
for item in prepared_data:
similarity = cosine_similarity(
query_embedding,
item["embedding"]
)
scores.append((similarity, item))
# --------------------------------------------------------
# SORT BY SIMILARITY
# --------------------------------------------------------
scores = sorted(
scores,
key=lambda x: x[0],
reverse=True
)
# --------------------------------------------------------
# RE-RANK USING KEYWORD BONUS
# --------------------------------------------------------
reranked = []
query_words = query.lower().split()
for sim, item in scores:
text = item["text"].lower()
keyword_bonus = sum(
word in text for word in query_words
)
final_score = sim + (0.03 * keyword_bonus)
reranked.append((final_score, item))
# --------------------------------------------------------
# SORT AGAIN AFTER RE-RANKING
# --------------------------------------------------------
reranked = sorted(
reranked,
key=lambda x: x[0],
reverse=True
)
# --------------------------------------------------------
# FILTER LOW SCORES
# --------------------------------------------------------
filtered = [
x for x in reranked
if x[0] >= min_score
]
return filtered[:top_k]
# ============================================================
# STEP 12 — TEST RETRIEVAL
# ============================================================
test_queries = [
"Why is my pod crashing repeatedly?",
"How to debug Kubernetes logs?",
"What causes OOMKilled?",
"How do services work in Kubernetes?",
"Why is my container restarting repeatedly?"
]
for query in test_queries:
print("\n" + "=" * 80)
print(f"QUERY: {query}\n")
results = retrieve(query)
for score, item in results:
print(f"Score: {score:.4f}")
print(item["text"])
print("-" * 40)
# ============================================================
# STEP 13 — BUILD PROMPT
# ============================================================
def build_prompt(query, retrieved_chunks):
context = "\n\n".join(
[item["text"] for score, item in retrieved_chunks]
)
prompt = f"""
You are a Kubernetes expert.
Answer ONLY using the provided context.
If the answer is not present in the context,
say "I don't know".
Context:
{context}
Question:
{query}
Answer:
"""
return prompt
# ============================================================
# STEP 14 — GENERATE ANSWER USING GEMINI
# ============================================================
def generate_answer(prompt):
model = genai.GenerativeModel("gemini-3-flash-preview")
#model = genai.GenerativeModel("gemini-1.5-flash")
response = model.generate_content(prompt)
return response.text
# ============================================================
# STEP 15 — COMPLETE RAG PIPELINE
# ============================================================
def rag_pipeline(query, top_k=3):
# --------------------------------------------------------
# RETRIEVE RELEVANT CHUNKS
# --------------------------------------------------------
retrieved_chunks = retrieve(
query=query,
top_k=top_k
)
# --------------------------------------------------------
# BUILD PROMPT
# --------------------------------------------------------
prompt = build_prompt(
query,
retrieved_chunks
)
# --------------------------------------------------------
# GENERATE FINAL ANSWER
# --------------------------------------------------------
answer = generate_answer(prompt)
return answer, retrieved_chunks
# ============================================================
# STEP 16 — FINAL TEST
# ============================================================
query = "Why is my pod crashing?"
answer, sources = rag_pipeline(query)
print("\n" + "=" * 80)
print("\n" + query + "\n")
print("FINAL ANSWER:\n")
print(answer)
print("\n" + "=" * 80)
print("RETRIEVED SOURCES:\n")
for score, item in sources:
print(f"Score: {score:.4f}")
print(item["text"])
print("-" * 40)
for query in test_queries :
answer, sources = rag_pipeline(query)
print("\n" + "=" * 80)
print("\n" + query + "\n")
print("FINAL ANSWER:\n")
print(answer)
print("\n" + "=" * 80)
=====================================================================================
OUTPUT (specifically check at the end how changing the query from "Why is my pod crashing?"
to "Why is my pod crashing repeatedly? " results in changes in scores.
=====================================================================================
No comments:
Post a Comment