Sunday, 1 November 2026

Build an AI-Powered Blog Chatbot Using LangChain, RAG, OpenAI and ChromaDB – Complete Hands-on Project

 

1. Introduction

Generative AI and Large Language Models (LLMs) are changing how we access and interact with information.

Traditional chatbots generally depend on predefined questions, answers or information available within their training data.

However, what if we want to build an AI assistant that can understand our own technical articles, retrieve relevant information and answer questions using our personal knowledge base?

In this project, we will build an AI-powered chatbot that connects directly to a personal Blogger website and answers questions based on published blog articles.

We will use Retrieval-Augmented Generation (RAG), LangChain, OpenAI embeddings, GPT-4.1-mini and ChromaDB.

The application will automatically:

  1. Fetch published articles from a Blogger website using its RSS feed.

  2. Extract article titles and content.

  3. Convert articles into LangChain documents.

  4. Split documents into smaller text chunks.

  5. Generate vector embeddings using OpenAI.

  6. Store embeddings in ChromaDB.

  7. Retrieve relevant content based on user questions.

  8. Generate contextual answers using an LLM.

  9. Display the original article references.

Project results

During my implementation, the application successfully processed:

MetricResult
Blog posts retrieved276
Text chunks generated2,773
Vector databaseChromaDB
Embedding modeltext-embedding-3-small
LLMGPT-4.1-mini
FrameworkLangChain

2. Project Architecture

Before writing the code, let's understand the complete architecture.

High-level architecture

Blogger Website
|
v
Blogger RSS Feed
|
v
Python RSS Data Loader
|
v
BeautifulSoup HTML Content Extraction
|
v
LangChain Documents
|
v
Recursive Character Text Splitter
|
v
OpenAI Embeddings
|
v
ChromaDB Vector Database
|
v
Retriever
|
v
Relevant Document Chunks
|
v
GPT-4.1-mini
|
v
AI Generated Answer + Source References

Architecture components

1. Blogger RSS Feed

Provides access to published blog articles without manually downloading individual posts.

2. Python

Handles RSS fetching, content extraction, document processing and application execution.

3. LangChain

Provides the document loading, splitting, embedding, retrieval and LLM orchestration components.

4. OpenAI Embeddings

Converts text into numerical vector representations that can be searched semantically.

5. ChromaDB

Stores the vectors and associated document metadata.

6. Retriever

Finds the most relevant document chunks based on the user's question.

7. GPT-4.1-mini

Uses retrieved information to generate a contextual answer.

3. Technology Stack

We will use the following technologies.

TechnologyPurpose
Python 3.13Application development
LangChainRAG orchestration
OpenAI APILLM and embeddings
GPT-4.1-miniAnswer generation
text-embedding-3-smallVector embeddings
ChromaDBVector database
feedparserRSS feed parsing
BeautifulSoupHTML extraction
RequestsHTTP requests
python-dotenvAPI key management

4. Prerequisites

Before starting, ensure you have:

  • Windows 11 or Linux.

  • Python 3.10 or newer.

  • VS Code.

  • OpenAI API key with API access and billing configured.

  • Internet connectivity.

  • A public Blogger website.

For this project, we will use:

Blog URL:

https://rajguptaaws.blogspot.com/

5. Create the Project Directory

Create a project directory:

mkdir day6-rag
cd day6-rag

Create the following project structure:

day6-rag/
│
├── main.py
│
├── .env
│
├── requirements.txt
│
└── chroma_db/

Explanation:

  • main.py: Main Python application.

  • .env: Stores the OpenAI API key.

  • requirements.txt: Python dependencies.

  • chroma_db: Persistent vector database storage.

6. Create Python Virtual Environment

We will use an existing virtual environment called Old/.venv.

Run:

& "..\Old\.venv\Scripts\python.exe" --version

Install the required dependencies:

& "..\Old\.venv\Scripts\python.exe" -m pip install requests beautifulsoup4 python-dotenv feedparser langchain langchain-community langchain-core langchain-openai langchain-chroma langchain-text-splitters

Verify the installation:

& "..\Old\.venv\Scripts\python.exe" -m pip list

7. Configure OpenAI API Key

Create a file named .env in the project directory.

Add:

OPENAI_API_KEY=your_openai_api_key

Replace the placeholder with your actual OpenAI API key.

Important security precautions:

  • Never publish your API key.

  • Never upload .env to GitHub.

  • Never hardcode API keys inside Python source code.

  • Add .env to .gitignore.

8. Understand the Blogger RSS Feed

Instead of scraping the complete Blogger website, we will use its RSS feed.

Our feed URL:

https://rajguptaaws.blogspot.com/feeds/posts/default?alt=rss

The feed contains information about published blog posts.

Examples of information available:

  • Article title

  • Article URL

  • Published date

  • Article content or summary

We can access the RSS feed programmatically using Python.

We will use feedparser to read the feed and Requests to download it.

9. Complete Python Implementation

Create main.py.

We will implement the application step by step.

Step 1: Import required libraries

import os
import hashlib
import requests
import feedparser

from bs4 import BeautifulSoup
from dotenv import load_dotenv

from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

Explanation:

  • requests: Downloads RSS feed content.

  • feedparser: Parses Blogger RSS entries.

  • BeautifulSoup: Extracts readable text from HTML.

  • Document: Creates LangChain document objects.

  • RecursiveCharacterTextSplitter: Splits articles into chunks.

  • OpenAIEmbeddings: Generates vector embeddings.

  • Chroma: Stores and retrieves vectors.

  • ChatOpenAI: Generates AI answers.

Step 2: Load environment variables

load_dotenv()

if not os.getenv("OPENAI_API_KEY"):
    raise ValueError("OPENAI_API_KEY is missing from .env")

This loads the OpenAI API key from the .env file.

Step 3: Define Blogger configuration

BLOG_URL = "https://rajguptaaws.blogspot.com"

FEED_URL = BLOG_URL + "/feeds/posts/default"

PAGE_SIZE = 100

We will retrieve articles in batches of 100.

Step 4: Load blog posts

Create a function to retrieve published blog posts.

def load_blog_posts():

    documents = []

    start_index = 1

    seen_urls = set()

    while True:

        params = {
            "alt": "rss",
            "start-index": start_index,
            "max-results": PAGE_SIZE
        }

        print(f"Fetching blog posts from index {start_index}...")

        response = requests.get(
            FEED_URL,
            params=params,
            timeout=30,
            headers={"User-Agent": "Mozilla/5.0"}
        )

        response.raise_for_status()

        feed = feedparser.parse(response.content)

        entries = feed.entries

        if not entries:
            break

        for entry in entries:

            title = entry.get("title", "Untitled")

            article_url = entry.get("link", BLOG_URL)

            if article_url in seen_urls:
                continue

            seen_urls.add(article_url)

            if entry.get("content"):

                html_content = entry.content[0].value

            else:

                html_content = entry.get("summary", "")

            soup = BeautifulSoup(
                html_content,
                "html.parser"
            )

            content = soup.get_text(
                separator="\n",
                strip=True
            )

            if not content:
                continue

            published = entry.get("published", "")

            document = Document(

                page_content=f"""
Title: {title}

Published: {published}

Content:

{content}
""",

                metadata={
                    "title": title,
                    "source": article_url,
                    "published": published
                }
            )

            documents.append(document)

        if len(entries) < PAGE_SIZE:
            break

        start_index += PAGE_SIZE

    return documents

Explanation:

The load_blog_posts() function performs the following operations:

  1. Connects to the Blogger RSS feed.

  2. Retrieves articles in batches.

  3. Extracts the article title.

  4. Retrieves the article URL.

  5. Extracts article HTML content.

  6. Converts HTML into plain text.

  7. Creates LangChain Document objects.

  8. Stores article metadata.

We use pagination because a blog may contain hundreds of articles.

Step 5: Load the blog documents

documents = load_blog_posts()

print(
    f"\nTotal blog posts loaded: {len(documents)}"
)

if not documents:
    raise RuntimeError(
        "No blog posts found. Check the RSS feed."
    )

Expected output from my implementation:

Fetching blog posts from index 1...
Fetching blog posts from index 101...
Fetching blog posts from index 201...

Total blog posts loaded: 276

The actual count depends on the number of published articles.

Step 6: Split documents into smaller chunks

Large articles should not be passed directly to the retrieval process as a single document.

We divide them into smaller chunks.

splitter = RecursiveCharacterTextSplitter(

    chunk_size=500,

    chunk_overlap=100
)

chunks = splitter.split_documents(documents)

print(
    f"Total chunks created: {len(chunks)}"
)

Explanation:

chunk_size=500

Each chunk contains approximately 500 characters.

chunk_overlap=100

Each consecutive chunk overlaps by approximately 100 characters.

Why do we need chunking?

  • Improves retrieval granularity.

  • Helps retrieve relevant sections.

  • Preserves some context between adjacent chunks.

  • Avoids sending entire articles to the LLM unnecessarily.

Expected output:

Total chunks created: 2773

Step 7: Initialize OpenAI embeddings

Embeddings convert text into numerical vector representations.

Initialize the embedding model:

embeddings = OpenAIEmbeddings(

    model="text-embedding-3-small"

)

Example:

Input:

"What is Kubernetes?"

The embedding model converts the text into a vector representation.

The resulting vector represents semantic information about the text.

We can use these vectors to find semantically related content.

Step 8: Initialize ChromaDB

ChromaDB is our vector database.

We will store the generated vectors and their corresponding document content.

vector_store = Chroma(

    collection_name="raj_blog_rag",

    embedding_function=embeddings,

    persist_directory="./chroma_db"

)

Explanation:

  • collection_name: Logical collection for our blog knowledge.

  • embedding_function: Converts documents into vectors.

  • persist_directory: Stores the database on local disk.

This allows us to reuse the vector database between application executions.

Step 9: Store documents in ChromaDB

We will create deterministic IDs for each chunk.

This prevents the same chunks from being repeatedly added with new random IDs.

chunk_ids = []

for chunk in chunks:

    unique_string = (
        chunk.metadata["source"]
        + "_"
        + chunk.page_content
    )

    chunk_id = hashlib.sha256(
        unique_string.encode("utf-8")
    ).hexdigest()

    chunk_ids.append(chunk_id)

Now insert the documents:

vector_store.add_documents(

    documents=chunks,

    ids=chunk_ids

)

print(
    "Blog data stored in ChromaDB successfully!"
)

Note: For this initial tutorial, the code demonstrates ingestion into a persistent collection. For repeated production ingestion, implement collection synchronization, updating changed articles and deleting outdated chunks. Do not assume that deterministic IDs alone automatically synchronize existing records.

Step 10: Create the retriever

The retriever searches ChromaDB for relevant document chunks.

retriever = vector_store.as_retriever(

    search_kwargs={
        "k": 4
    }

)

The k=4 parameter tells the retriever to return four relevant chunks for each question.

Step 11: Initialize the LLM

We will use GPT-4.1-mini.

llm = ChatOpenAI(

    model="gpt-4.1-mini",

    temperature=0

)

Temperature 0 helps make responses more consistent.

Step 12: Create the RAG prompt

The prompt instructs the LLM to answer using the retrieved blog content.

prompt = ChatPromptTemplate.from_template("""

You are an expert AI assistant for Raj Gupta's
technical blog.

Answer the user's question using the provided context.

Rules:

1. Use the retrieved blog content.
2. Do not invent technical information.
3. If the answer is unavailable, say you do not know.
4. Explain technical concepts clearly.
5. Include relevant source URLs when available.

Context:

{context}

Question:

{question}

Answer:

""")

Step 13: Format retrieved documents

We need to combine retrieved document chunks into a single context string.

def format_docs(docs):

    formatted_docs = []

    for doc in docs:

        content = f"""

Title: {doc.metadata.get('title', '')}

Source: {doc.metadata.get('source', '')}

Content:

{doc.page_content}

"""

        formatted_docs.append(content)

    return "\n\n".join(formatted_docs)

This function also preserves the source URL so the LLM can reference the original article.

Step 14: Implement the RAG question-answering function

We will retrieve relevant documents first and then send them to the LLM.

def ask_question(question):

    # Retrieve relevant documents

    retrieved_docs = retriever.invoke(question)

    if not retrieved_docs:

        return "I could not find relevant information."

    # Format retrieved documents

    context = format_docs(retrieved_docs)

    # Create prompt

    messages = prompt.invoke({

        "context": context,

        "question": question

    })

    # Generate answer

    response = llm.invoke(messages)

    answer = StrOutputParser().invoke(response)

    # Display source references

    print("\nAI Answer:\n")

    print(answer)

    print("\nSources:")

    sources = set()

    for doc in retrieved_docs:

        source = doc.metadata.get("source")

        if source:

            sources.add(source)

    for source in sources:

        print(source)

This approach prints sources directly from the retrieved documents instead of relying entirely on the LLM to reproduce URLs accurately.

Step 15: Create the interactive chatbot

Finally, create an interactive terminal interface.

print("\n====================================")

print("Raj Gupta Blog AI Assistant")

print("====================================")


while True:

    question = input(
        "\nAsk a question (exit to stop): "
    )

    if question.lower().strip() == "exit":

        print("Exiting chatbot...")

        break

    if not question.strip():

        continue

    try:

        ask_question(question)

    except Exception as error:

        print(f"\nError: {error}")

10. Complete main.py

All the previous steps are combined into one Python application.

The complete file contains:

  • RSS feed ingestion

  • Document extraction

  • Text chunking

  • Embedding generation

  • ChromaDB storage

  • Semantic retrieval

  • LLM integration

  • Interactive question answering

  • Source reference display

Use the complete code in the accompanying implementation section of this article.

11. Run the Application

Open PowerShell inside the day6-rag directory.

Execute:

& "..\Old\.venv\Scripts\python.exe" main.py

The application will start fetching blog posts.

Expected output:

Fetching blog posts from index 1...
Fetching blog posts from index 101...
Fetching blog posts from index 201...

Total blog posts loaded: 276

Total chunks created: 2773

Blog data stored in ChromaDB successfully!

====================================
Raj Gupta Blog AI Assistant
====================================

Ask a question (exit to stop):

12. Test the AI Chatbot

Let's test the application with a real question.

Question:

What is RAG?

The chatbot retrieves relevant information from the blog and passes it to GPT-4.1-mini.

Example answer:

RAG stands for Retrieval-Augmented Generation.

It is a method used to improve the responses of a language model by incorporating relevant external information retrieved from a knowledge base or documents.

A typical RAG process is:

User Question → Document Retrieval → Context → LLM → Answer

The chatbot also displays the original blog article reference.

Example source:

https://rajguptaaws.blogspot.com/2026/10/part-2-building-production-style-rag.html

13. How RAG Works Internally

Let's understand the two important phases.

Phase 1: Data Ingestion

This happens before users ask questions.

  1. Fetch blog articles.

  2. Extract article content.

  3. Create document objects.

  4. Split articles into chunks.

  5. Generate embeddings.

  6. Store vectors in ChromaDB.

Phase 2: Query Execution

This happens whenever a user asks a question.

  1. Receive the user question.

  2. Convert the question into an embedding.

  3. Search ChromaDB for similar vectors.

  4. Retrieve the most relevant document chunks.

  5. Construct the prompt with retrieved context.

  6. Send the prompt to GPT-4.1-mini.

  7. Generate the final answer.

  8. Display relevant source URLs.

14. Common Errors and Troubleshooting

Error 1: ModuleNotFoundError: No module named 'feedparser'

Solution:

python -m pip install feedparser

Error 2: ModuleNotFoundError: No module named 'bs4'

Solution:

python -m pip install beautifulsoup4

Error 3: ModuleNotFoundError: No module named 'langchain_chroma'

Solution:

python -m pip install langchain-chroma

Error 4: OpenAI API authentication error

Check whether the API key is correctly configured in the .env file.

Also verify that your OpenAI API account has access and available billing.

Error 5: Blogger RSS feed returns no articles

Verify:

  • Blog is publicly accessible.

  • RSS feed URL is correct.

  • Blogger feed settings permit access.

  • The feed is returning article entries.

Error 6: ChromaDB retrieval returns irrelevant answers

Possible improvements:

  • Experiment with chunk size.

  • Experiment with chunk overlap.

  • Increase the number of retrieved chunks.

  • Improve document metadata.

  • Use a reranker.

  • Evaluate retrieval quality.

15. Production Improvements

The current implementation is a learning-level RAG application.

For a production-grade implementation, consider:

  1. Incremental ingestion instead of processing all articles every time.

  2. Detecting modified and deleted articles.

  3. Metadata filtering.

  4. Hybrid search using keyword and vector retrieval.

  5. Reranking retrieved documents.

  6. Query rewriting.

  7. RAG evaluation using Ragas.

  8. FastAPI integration.

  9. Authentication and authorization.

  10. Logging, monitoring and error handling.

  11. Rate limiting and API cost management.

  12. Deployment using Docker and Kubernetes.

16. Future Enhancements

The next version of this project can include:

  • FastAPI backend.

  • Web-based chatbot UI.

  • LangGraph workflow orchestration.

  • Conversation memory.

  • Multi-user support.

  • Automated blog synchronization.

  • Advanced retrieval techniques.

  • MCP server integration.

  • AI agents for technical troubleshooting.

17. Conclusion

In this project, we successfully built an AI-powered chatbot that uses our own technical blog as its knowledge base.

We explored how Blogger RSS feeds, Python, LangChain, OpenAI embeddings, ChromaDB and GPT-4.1-mini work together to implement a complete Retrieval-Augmented Generation pipeline.

The project demonstrates how an LLM can access custom knowledge beyond its pretrained information and generate answers grounded in retrieved documents.

This is a practical foundation for building enterprise knowledge assistants, internal documentation chatbots, technical support agents and AI-powered information retrieval systems.

Author: Raj Gupta

Technologies: Python | LangChain | RAG | OpenAI | ChromaDB | Generative AI | Vector Database | LLM

Project: AI-Powered Blog Chatbot


Wednesday, 7 October 2026

Part 2 — Building a Production-Style RAG Pipeline for a Telecom AI Support Agent

Building a Production-Style RAG Pipeline for a Telecom AI Support Agent — Part 2

Introduction

In Part 1 of the GlobalNet Support Agent project, I built an AI-powered telecom customer support application using:

  • Python

  • FastAPI

  • Google Cloud

  • Vertex AI

  • Gemini

  • Multiple AI agents

The application already contained three agents:

Customer
   ↓
Upsell Agent
   ↓
Plan Agent
   ↓
Offer Agent
   ↓
Customer Response

The Upsell Agent determines whether an upgrade is justified.

The Plan Agent determines which plan is suitable.

The Offer Agent creates a personalized customer offer.

However, there was an important limitation.

The Plan Agent initially contained plan information directly inside its prompt:

BASIC     → 100 Mbps
STANDARD  → 300 Mbps
PREMIUM   → 500 Mbps
ULTRA     → 1 Gbps

This approach works for a prototype, but it does not scale.

In this part, I will convert the application into an Agent + RAG architecture using:

JSON Knowledge Base
        ↓
Document Processing
        ↓
Chunking
        ↓
Sentence Transformers
        ↓
Embeddings
        ↓
ChromaDB
        ↓
Semantic Search
        ↓
Relevance Filtering
        ↓
Gemini
        ↓
Plan Recommendation

We will also build a basic RAG evaluation framework to test whether the retriever is actually finding the correct telecom plans.


1. What is RAG?

RAG stands for:

Retrieval-Augmented Generation

A normal LLM application might work like this:

User Question
      ↓
Prompt
      ↓
LLM
      ↓
Answer

The problem is that the LLM may not know our private business information.

For example, Gemini does not automatically know the current internal GlobalNet plan catalog used by this demo application.

We could put all plan information directly into the prompt, but this becomes difficult as the knowledge base grows.

RAG changes the architecture.

User Question
      ↓
Retriever
      ↓
Knowledge Base
      ↓
Relevant Information
      ↓
LLM
      ↓
Grounded Answer

Instead of expecting the LLM to know everything, we retrieve relevant information first and provide that information as context.


2. Why RAG for a Telecom Support Agent?

Imagine that a telecom company has:

100 internet plans
500 support documents
200 troubleshooting guides
50 promotion documents
1000 FAQ entries
Product documentation
Network policies
Upgrade rules
Eligibility rules

Sending all of this information to an LLM for every customer request would be inefficient.

Instead:

Customer:
"I have 8 devices and multiple 4K TVs."
                 ↓
             Retriever
                 ↓
       Search knowledge base
                 ↓
      Relevant plan information
                 ↓
              Gemini
                 ↓
        Recommended plan

This is where RAG becomes useful.


3. RAG Architecture for GlobalNet

The architecture developed in this project is:

                         Customer
                            |
                            v
                      FastAPI API
                            |
                            v
                     Upsell Agent
                            |
                            v
                         Gemini
                            |
                            v
                      Plan Agent
                            |
                            v
                       RAG Query
                            |
                            v
                 Sentence Transformer
                            |
                            v
                     Query Embedding
                            |
                            v
                        ChromaDB
                            |
                            v
                  Similarity Search
                            |
                            v
                    Relevant Chunks
                            |
                            v
                  Relevance Filtering
                            |
                            v
                     Context Prompt
                            |
                            v
                         Gemini
                            |
                            v
                 Plan Recommendation
                            |
                            v
                      Offer Agent
                            |
                            v
                         Gemini
                            |
                            v
                  Personalized Offer

4. Updated Project Directory Structure

After adding RAG, the project structure becomes:

GlobalNet-Support-Agent-GCP/
│
├── app/
│   ├── __init__.py
│   ├── main.py
│   │
│   ├── agents/
│   │   ├── __init__.py
│   │   ├── upsell_agent.py
│   │   ├── plan_agent.py
│   │   └── offer_agent.py
│   │
│   ├── services/
│   │   ├── __init__.py
│   │   ├── vertex_ai.py
│   │   ├── plan_catalog.py
│   │   ├── rag.py
│   │   ├── firestore.py
│   │   └── email.py
│   │
│   └── models/
│       └── customer.py
│
├── data/
│   ├── plans/
│   │   └── globalnet_plans.json
│   │
│   └── rag_evaluation.json
│
├── scripts/
│   ├── ingest_plans.py
│   ├── evaluate_rag.py
│   └── evaluate_rag_top1.py
│
├── chroma_db/
│
├── test_vertex.py
├── test_rag.py
│
├── requirements.txt
├── Dockerfile
├── .dockerignore
└── README.md

5. Create the GlobalNet Knowledge Base

The first step is moving plan information outside the Python source code.

Create:

data/plans/globalnet_plans.json

Example:

[
  {
    "plan_id": "BASIC",
    "plan_name": "GlobalNet Basic",
    "speed": "100 Mbps",
    "monthly_price": 499,
    "data_limit": "Unlimited",
    "recommended_for": [
      "1-3 devices",
      "normal browsing",
      "email",
      "social media",
      "HD streaming"
    ],
    "features": [
      "Unlimited data",
      "Standard customer support",
      "Wi-Fi router"
    ]
  },
  {
    "plan_id": "STANDARD",
    "plan_name": "GlobalNet Standard",
    "speed": "300 Mbps",
    "monthly_price": 799,
    "data_limit": "Unlimited",
    "recommended_for": [
      "3-6 devices",
      "4K streaming",
      "work from home",
      "video conferencing",
      "online gaming"
    ],
    "features": [
      "Unlimited data",
      "Priority customer support",
      "Wi-Fi 6 router"
    ]
  },
  {
    "plan_id": "PREMIUM",
    "plan_name": "GlobalNet Premium",
    "speed": "500 Mbps",
    "monthly_price": 1099,
    "data_limit": "Unlimited",
    "recommended_for": [
      "6-10 devices",
      "multiple 4K streams",
      "heavy work from home",
      "online gaming",
      "large file transfers"
    ],
    "features": [
      "Unlimited data",
      "Priority customer support",
      "Wi-Fi 6 router",
      "Premium support"
    ]
  },
  {
    "plan_id": "ULTRA",
    "plan_name": "GlobalNet Ultra",
    "speed": "1 Gbps",
    "monthly_price": 1499,
    "data_limit": "Unlimited",
    "recommended_for": [
      "10+ devices",
      "multiple simultaneous 4K streams",
      "professional users",
      "heavy cloud workloads",
      "large household usage"
    ],
    "features": [
      "Unlimited data",
      "24x7 priority support",
      "Wi-Fi 6 router",
      "Premium support",
      "Dedicated service assistance"
    ]
  }
]

These are sample plans created for this learning project.

They are not intended to represent actual commercial telecom offerings.


6. Create a Plan Catalog Service

Create:

app/services/plan_catalog.py

Code:

import json

from pathlib import Path


PLAN_FILE = (
    Path(__file__).resolve().parents[2]
    / "data"
    / "plans"
    / "globalnet_plans.json"
)


def load_plans():

    with open(
        PLAN_FILE,
        "r",
        encoding="utf-8"
    ) as file:

        return json.load(file)


def get_plan(plan_id: str):

    plans = load_plans()

    for plan in plans:

        if plan["plan_id"] == plan_id:

            return plan

    return None

This separates business data from application logic.

The application can now load plan information from:

globalnet_plans.json

instead of embedding the information inside Python prompts.


7. Moving from JSON Lookup to RAG

Simply reading JSON is not RAG.

This:

JSON
 ↓
Read everything
 ↓
Put everything into prompt
 ↓
Gemini

is still a normal context-injection approach.

True RAG introduces retrieval:

Documents
    ↓
Chunking
    ↓
Embeddings
    ↓
Vector Database

         +

Customer Question
    ↓
Embedding
    ↓
Vector Search
    ↓
Relevant Documents
    ↓
Gemini

8. Install the RAG Components

For this project I used:

ChromaDB
Sentence Transformers

Install:

pip install chromadb sentence-transformers

Verify:

python -c "import chromadb; import sentence_transformers; print('RAG packages OK')"

9. What is an Embedding?

An embedding converts text into a numerical representation.

For example:

"I have many devices and multiple 4K TVs."

is converted conceptually into:

[0.13, -0.44, 0.82, 0.07, ...]

The actual embedding contains many numerical dimensions.

Another sentence with similar meaning should have a vector located relatively close in embedding space.

For example:

"I need high-speed internet for several devices and 4K streaming."

has similar semantic meaning.

This enables semantic search.


10. Embedding Model

The project uses:

SentenceTransformer(
    "all-MiniLM-L6-v2"
)

This model runs locally and converts plan documents and customer questions into embeddings.


11. Why Use a Vector Database?

A traditional database can perform queries such as:

plan_id = "PREMIUM"

But a customer does not necessarily know the plan ID.

The customer might say:

"I have 8 devices and my family watches multiple 4K streams."

We need to find the plan whose meaning best matches this requirement.

A vector database stores embeddings and performs similarity searches.

For this project I used:

ChromaDB

12. Why Chunking Matters

Initially I stored one entire plan as one document.

For example:

PREMIUM
  |
  +-- Price
  +-- Speed
  +-- Recommended Usage
  +-- Features

That works with a tiny dataset.

However, larger knowledge bases require more granular retrieval.

I therefore split each plan into three chunks.

PREMIUM
   |
   +---- PREMIUM_basic
   |
   +---- PREMIUM_usage
   |
   +---- PREMIUM_features

The same happens for all four plans.

BASIC_basic
BASIC_usage
BASIC_features

STANDARD_basic
STANDARD_usage
STANDARD_features

PREMIUM_basic
PREMIUM_usage
PREMIUM_features

ULTRA_basic
ULTRA_usage
ULTRA_features

Therefore:

4 plans × 3 chunks = 12 chunks

Each chunk receives its own embedding.


13. Complete RAG Service

Create:

app/services/rag.py

Code:

import chromadb

from sentence_transformers import SentenceTransformer


CHROMA_PATH = "./chroma_db"

COLLECTION_NAME = "globalnet_plans"


# ------------------------------------------------------------
# Embedding Model
# ------------------------------------------------------------

embedding_model = SentenceTransformer(
    "all-MiniLM-L6-v2"
)


# ------------------------------------------------------------
# ChromaDB Client
# ------------------------------------------------------------

chroma_client = chromadb.PersistentClient(
    path=CHROMA_PATH
)


# ------------------------------------------------------------
# ChromaDB Collection
# ------------------------------------------------------------

collection = chroma_client.get_or_create_collection(
    name=COLLECTION_NAME
)


# ------------------------------------------------------------
# Create Plan Chunks
# ------------------------------------------------------------

def create_plan_chunks(plans):

    documents = []
    ids = []
    metadatas = []

    for plan in plans:

        plan_id = plan["plan_id"]

        # Basic information chunk

        basic_info = f"""
Plan ID: {plan_id}

Plan Name: {plan["plan_name"]}

Speed: {plan["speed"]}

Monthly Price: ₹{plan["monthly_price"]}

Data Limit: {plan["data_limit"]}
"""

        documents.append(
            basic_info
        )

        ids.append(
            f"{plan_id}_basic"
        )

        metadatas.append({
            "plan_id": plan_id,
            "plan_name": plan["plan_name"],
            "chunk_type": "basic",
            "speed": plan["speed"],
        })


        # Recommended usage chunk

        usage_info = f"""
Plan ID: {plan_id}

Plan Name: {plan["plan_name"]}

Recommended For:

{", ".join(plan["recommended_for"])}
"""

        documents.append(
            usage_info
        )

        ids.append(
            f"{plan_id}_usage"
        )

        metadatas.append({
            "plan_id": plan_id,
            "plan_name": plan["plan_name"],
            "chunk_type": "usage",
            "speed": plan["speed"],
        })


        # Features chunk

        feature_info = f"""
Plan ID: {plan_id}

Plan Name: {plan["plan_name"]}

Features:

{", ".join(plan["features"])}
"""

        documents.append(
            feature_info
        )

        ids.append(
            f"{plan_id}_features"
        )

        metadatas.append({
            "plan_id": plan_id,
            "plan_name": plan["plan_name"],
            "chunk_type": "features",
            "speed": plan["speed"],
        })

    return documents, ids, metadatas


# ------------------------------------------------------------
# Create Embeddings
# ------------------------------------------------------------

def create_embeddings(documents):

    embeddings = embedding_model.encode(
        documents
    )

    return embeddings.tolist()


# ------------------------------------------------------------
# Store Plans
# ------------------------------------------------------------

def store_plans(plans):

    documents, ids, metadatas = create_plan_chunks(
        plans
    )

    embeddings = create_embeddings(
        documents
    )

    collection.upsert(
        ids=ids,
        documents=documents,
        embeddings=embeddings,
        metadatas=metadatas,
    )

    return len(documents)


# ------------------------------------------------------------
# Search Plans
# ------------------------------------------------------------

def search_plans(
    query,
    top_k=5,
    max_distance=0.80
):

    query_embedding = embedding_model.encode(
        [query]
    ).tolist()

    results = collection.query(
        query_embeddings=query_embedding,
        n_results=top_k,
        include=[
            "documents",
            "metadatas",
            "distances"
        ],
    )

    filtered_documents = []
    filtered_metadatas = []
    filtered_distances = []

    documents = results.get(
        "documents",
        [[]]
    )[0]

    metadatas = results.get(
        "metadatas",
        [[]]
    )[0]

    distances = results.get(
        "distances",
        [[]]
    )[0]

    for document, metadata, distance in zip(
        documents,
        metadatas,
        distances,
    ):

        if distance <= max_distance:

            filtered_documents.append(
                document
            )

            filtered_metadatas.append(
                metadata
            )

            filtered_distances.append(
                distance
            )

    return {
        "documents": [
            filtered_documents
        ],
        "metadatas": [
            filtered_metadatas
        ],
        "distances": [
            filtered_distances
        ],
    }

14. What Happens During Ingestion?

The ingestion pipeline works like this:

globalnet_plans.json
        ↓
load_plans()
        ↓
4 Plans
        ↓
create_plan_chunks()
        ↓
12 Chunks
        ↓
SentenceTransformer
        ↓
12 Embeddings
        ↓
ChromaDB

This happens before customer queries are processed.


15. Create the Ingestion Script

Create:

scripts/ingest_plans.py

Code:

import sys

from pathlib import Path


PROJECT_ROOT = Path(__file__).resolve().parents[1]

sys.path.insert(
    0,
    str(PROJECT_ROOT)
)


from app.services.plan_catalog import load_plans
from app.services.rag import store_plans


def main():

    print(
        "Loading GlobalNet plans..."
    )

    plans = load_plans()

    print(
        f"Found {len(plans)} plans"
    )

    print(
        "Creating embeddings and storing in ChromaDB..."
    )

    count = store_plans(
        plans
    )

    print(
        f"Successfully stored {count} chunks in ChromaDB"
    )


if __name__ == "__main__":

    main()

Run:

python scripts/ingest_plans.py

Expected:

Loading GlobalNet plans...

Found 4 plans

Creating embeddings and storing in ChromaDB...

Successfully stored 12 chunks in ChromaDB

16. What Gets Stored in ChromaDB?

Each chunk contains three important things:

ID
Document
Metadata
Embedding

For example:

ID:
PREMIUM_usage

Document:

Plan ID: PREMIUM

Plan Name: GlobalNet Premium

Recommended For:

6-10 devices,
multiple 4K streams,
heavy work from home,
online gaming,
large file transfers

Metadata:

{
  "plan_id": "PREMIUM",
  "plan_name": "GlobalNet Premium",
  "chunk_type": "usage",
  "speed": "500 Mbps"
}

And internally:

Embedding Vector

This makes the knowledge searchable semantically.


17. Query-Time RAG Flow

Suppose the customer says:

"I have eight devices and use multiple 4K streams."

The system does not search for an exact string.

Instead:

Customer Question
       ↓
SentenceTransformer
       ↓
Query Embedding
       ↓
ChromaDB
       ↓
Vector Similarity Search
       ↓
Closest Chunks

A likely result could include:

PREMIUM_usage

because that document contains semantically related information:

6-10 devices
multiple 4K streams

18. Testing RAG Search

Create:

test_rag.py

Code:

from app.services.rag import search_plans


query = """
I have many devices at home and use several
4K streams. I need high bandwidth.
"""


results = search_plans(
    query,
    top_k=5
)


print(
    "\n===== RAG RESULTS =====\n"
)


documents = results["documents"][0]

metadatas = results["metadatas"][0]

distances = results["distances"][0]


for i in range(
    len(documents)
):

    print(
        f"Result: {i + 1}"
    )

    print(
        f"Plan: {metadatas[i]['plan_name']}"
    )

    print(
        f"Chunk Type: {metadatas[i]['chunk_type']}"
    )

    print(
        f"Distance: {distances[i]}"
    )

    print(
        "\nDocument:"
    )

    print(
        documents[i]
    )

    print(
        "-" * 70
    )

Run:

python test_rag.py

19. What is top_k?

Our query contains:

top_k=5

This means:

Retrieve up to the 5 nearest chunks.

Conceptually:

Customer Query
      ↓
Vector Search
      ↓
Result 1
Result 2
Result 3
Result 4
Result 5

The application then decides which results are relevant enough to use.


20. Understanding Distance

ChromaDB returns distance information for retrieved vectors.

For our chosen setup, smaller distances generally represent greater similarity.

For example:

Document A
Distance = 0.25

Document B
Distance = 0.70

Document A is generally a closer semantic match.

However, distance values should not be interpreted as universal percentages.

A value such as:

0.80

does not mean:

80% relevant

The values depend on the embedding model, vector space and configured distance behavior.


21. Relevance Filtering

A vector database normally returns the nearest available documents even when none are especially useful.

Suppose the user asks:

"What is the weather today?"

Our database contains only telecom plans.

Without filtering, ChromaDB may still return the closest telecom document.

We don't want Gemini to treat that as valid knowledge.

Therefore, the RAG service applies:

if distance <= max_distance:

The current learning-project threshold is:

max_distance=0.80

Results outside the threshold are rejected.

This creates:

Query
   ↓
Vector Search
   ↓
Candidate Documents
   ↓
Distance Filter
   ↓
Relevant Documents

22. Why Relevance Thresholds Need Evaluation

The value:

0.80

is not automatically correct for every RAG application.

The best threshold depends on:

Embedding model
Document type
Chunk size
Domain
Query style
Vector database configuration
Evaluation dataset

Therefore, a production system should determine the threshold using evaluation rather than guessing.


23. Connect RAG to the Plan Agent

The Plan Agent previously received every plan.

Now it retrieves only relevant plan information.

File:

app/agents/plan_agent.py

Code:

from app.services.vertex_ai import generate_text

from app.services.rag import search_plans


def recommend_plan(
    customer,
    upsell_analysis: str
) -> str:

    query = f"""
Customer currently has:

Plan:
{customer.current_plan}

Customer problem:
{customer.chat_text}

Find the GlobalNet plan information that is
most relevant to the customer's requirements.
"""

    rag_results = search_plans(
        query,
        top_k=5,
        max_distance=0.80
    )

    documents = (
        rag_results["documents"][0]
    )

    distances = (
        rag_results["distances"][0]
    )

    if not documents:

        return """
Current Plan:
Unknown

Recommended Plan:
No recommendation

Reason:
No sufficiently relevant plan information was found
in the GlobalNet knowledge base.

Next Action:
Ask a support specialist to review the customer.
"""

    relevant_context = ""

    for document, distance in zip(
        documents,
        distances
    ):

        relevant_context += f"""
Relevance Distance: {distance}

{document}

-----------------------------
"""

    prompt = f"""
You are a telecom plan recommendation AI.

Customer Information:

Customer ID:
{customer.customer_id}

Customer Name:
{customer.customer_name}

Current Plan:
{customer.current_plan}

Current Plan Description:
{customer.current_plan_desc}

Loyalty Status:
{customer.loyalty_status}

Tenure:
{customer.tenure_months} months

Customer Message:

{customer.chat_text}

Upsell Analysis:

{upsell_analysis}

Relevant GlobalNet Knowledge:

{relevant_context}

Rules:

1. Use only information contained in the retrieved
   GlobalNet knowledge.

2. Never invent a plan.

3. Never invent a price.

4. Never invent a speed.

5. Do not recommend an upgrade just to increase revenue.

6. If a technical issue is likely,
   recommend troubleshooting.

7. If an upgrade is justified,
   recommend the lowest suitable plan.

8. Consider the customer's actual usage.

9. Clearly explain the recommendation.

Return:

Current Plan:
Recommended Plan:
Speed:
Monthly Price:
Reason:
Next Action:
"""

    return generate_text(
        prompt
    )

24. Agent + RAG Architecture

The Plan Agent now works as:

Customer
   ↓
Plan Agent
   ↓
Build Retrieval Query
   ↓
Sentence Transformer
   ↓
Query Embedding
   ↓
ChromaDB
   ↓
Top-K Chunks
   ↓
Distance Filter
   ↓
Relevant Knowledge
   ↓
Gemini Prompt
   ↓
Plan Recommendation

This is significantly different from simply asking an LLM:

"Which plan should I recommend?"

The LLM now receives information retrieved from our own knowledge base.


25. Why Grounding Matters

Without RAG:

Customer
   ↓
Gemini
   ↓
Possible answer based on model knowledge

With RAG:

Customer
   ↓
GlobalNet Knowledge
   ↓
Relevant Plans
   ↓
Gemini
   ↓
Grounded recommendation

We also explicitly instruct Gemini:

Never invent a plan.
Never invent a price.
Never invent a speed.

RAG does not mathematically guarantee that hallucinations disappear, but retrieval plus explicit grounding rules can substantially improve control over the information available to the model.


26. Testing the Complete API

Start:

uvicorn app.main:app --reload

Open:

http://127.0.0.1:8000/docs

Test:

POST /support

Example:

{
  "customer_id": "1001",
  "chat_text": "My internet is very slow and I need much higher bandwidth because I have many devices and use 4K streaming.",
  "loyalty_status": "GOLD",
  "current_plan": "BASIC",
  "current_plan_desc": "100 Mbps",
  "tenure_months": 48,
  "customer_email": "your-email@example.com",
  "customer_name": "Raj",
  "call_type": "online"
}

The execution path is now:

POST /support
      ↓
Upsell Agent
      ↓
Gemini
      ↓
Plan Agent
      ↓
RAG
      ↓
ChromaDB
      ↓
Relevant Plans
      ↓
Gemini
      ↓
Plan Recommendation
      ↓
Offer Agent
      ↓
Gemini
      ↓
Final Response

27. Why RAG Evaluation is Necessary

Getting a response from RAG does not mean RAG is good.

We need to test:

Did RAG retrieve the correct information?

For example:

Question:
"I have 8 devices and multiple 4K streams."

Expected:
PREMIUM

If RAG returns only:

BASIC

our system has a retrieval problem.

This problem exists before Gemini even generates an answer.

Therefore, retrieval should be evaluated independently.


28. Create a RAG Evaluation Dataset

Create:

data/rag_evaluation.json

Example:

[
  {
    "question": "I only have two devices and mainly browse websites and use email.",
    "expected_plan": "BASIC"
  },
  {
    "question": "I have four devices and frequently work from home with video conferencing.",
    "expected_plan": "STANDARD"
  },
  {
    "question": "I have eight devices and use multiple 4K streams.",
    "expected_plan": "PREMIUM"
  },
  {
    "question": "I have more than ten devices and several simultaneous 4K streams.",
    "expected_plan": "ULTRA"
  },
  {
    "question": "I need internet for normal browsing, email and social media.",
    "expected_plan": "BASIC"
  },
  {
    "question": "My family has many devices and we regularly play online games and stream 4K video.",
    "expected_plan": "PREMIUM"
  }
]

This is not training data.

It is a test dataset.


29. RAG Evaluation Script

Create:

scripts/evaluate_rag.py

Code:

import json

from pathlib import Path

from app.services.rag import search_plans


PROJECT_ROOT = (
    Path(__file__).resolve().parents[1]
)


EVALUATION_FILE = (
    PROJECT_ROOT
    / "data"
    / "rag_evaluation.json"
)


def load_evaluation_data():

    with open(
        EVALUATION_FILE,
        "r",
        encoding="utf-8"
    ) as file:

        return json.load(file)


def extract_plan_ids(results):

    plan_ids = []

    for metadata in (
        results["metadatas"][0]
    ):

        plan_id = metadata["plan_id"]

        if plan_id not in plan_ids:

            plan_ids.append(
                plan_id
            )

    return plan_ids


def evaluate():

    test_cases = load_evaluation_data()

    total = len(
        test_cases
    )

    correct = 0

    print(
        "\n===== RAG EVALUATION =====\n"
    )

    for index, test_case in enumerate(
        test_cases,
        start=1
    ):

        question = (
            test_case["question"]
        )

        expected_plan = (
            test_case["expected_plan"]
        )

        results = search_plans(
            question,
            top_k=5,
            max_distance=0.80
        )

        retrieved_plans = extract_plan_ids(
            results
        )

        if expected_plan == "NONE":

            is_correct = (
                len(retrieved_plans) == 0
            )

        else:

            is_correct = (
                expected_plan
                in retrieved_plans
            )

        if is_correct:

            correct += 1

        print(
            f"Test {index}"
        )

        print(
            f"Question: {question}"
        )

        print(
            f"Expected: {expected_plan}"
        )

        print(
            f"Retrieved: {retrieved_plans}"
        )

        print(
            f"Result: {'PASS' if is_correct else 'FAIL'}"
        )

        print(
            "-" * 70
        )

    accuracy = (
        correct / total
    ) * 100

    print(
        "\n===== SUMMARY ====="
    )

    print(
        f"Total Tests: {total}"
    )

    print(
        f"Correct: {correct}"
    )

    print(
        f"Incorrect: {total - correct}"
    )

    print(
        f"Accuracy: {accuracy:.2f}%"
    )


if __name__ == "__main__":

    evaluate()

30. Run RAG Evaluation

Run:

python scripts/evaluate_rag.py

Example output:

===== RAG EVALUATION =====

Test 1

Question:
I only have two devices and mainly browse websites and use email.

Expected:
BASIC

Retrieved:
['BASIC', 'STANDARD']

Result:
PASS

At the end:

===== SUMMARY =====

Total Tests: 6

Correct: 5

Incorrect: 1

Accuracy: 83.33%

The exact result depends on retrieval behavior.

A failed test is not something to hide.

It tells us where retrieval needs improvement.


31. Top-K Evaluation

Suppose RAG returns:

1. PREMIUM
2. STANDARD
3. ULTRA
4. BASIC

and the expected answer is:

PREMIUM

For a Top-K retrieval test:

PASS

because PREMIUM was retrieved.

This answers:

Did the retriever find the relevant information somewhere in its result set?


32. Top-1 Evaluation

A stricter test checks only the first result.

For example:

Question:
8 devices + multiple 4K streams

Expected:
PREMIUM

Top Result:
PREMIUM

Result:

PASS

But:

Top Result:
STANDARD

would be:

FAIL

Top-1 accuracy tells us how often the retriever's highest-ranked result is correct.


33. Create Top-1 Evaluation

Create:

scripts/evaluate_rag_top1.py

Code:

import json

from pathlib import Path

from app.services.rag import search_plans


PROJECT_ROOT = (
    Path(__file__).resolve().parents[1]
)


EVALUATION_FILE = (
    PROJECT_ROOT
    / "data"
    / "rag_evaluation.json"
)


def load_data():

    with open(
        EVALUATION_FILE,
        "r",
        encoding="utf-8"
    ) as file:

        return json.load(file)


def evaluate():

    test_cases = load_data()

    correct = 0

    total = len(
        test_cases
    )

    print(
        "\n===== TOP-1 RAG EVALUATION =====\n"
    )

    for index, test_case in enumerate(
        test_cases,
        start=1
    ):

        question = (
            test_case["question"]
        )

        expected = (
            test_case["expected_plan"]
        )

        results = search_plans(
            question,
            top_k=1,
            max_distance=0.80
        )

        if results["metadatas"][0]:

            retrieved = (
                results["metadatas"][0][0]["plan_id"]
            )

            distance = (
                results["distances"][0][0]
            )

        else:

            retrieved = None
            distance = None

        if expected == "NONE":

            passed = (
                retrieved is None
            )

        else:

            passed = (
                retrieved == expected
            )

        if passed:

            correct += 1

        print(
            f"Test: {index}"
        )

        print(
            f"Expected: {expected}"
        )

        print(
            f"Retrieved: {retrieved}"
        )

        print(
            f"Distance: {distance}"
        )

        print(
            f"Result: {'PASS' if passed else 'FAIL'}"
        )

        print(
            "-" * 60
        )

    accuracy = (
        correct / total
    ) * 100

    print(
        f"\nTop-1 Accuracy: {accuracy:.2f}%"
    )


if __name__ == "__main__":

    evaluate()

Run:

python scripts/evaluate_rag_top1.py

34. Testing Abstention

An important test is an unrelated question.

Add:

{
  "question": "What is the capital city of India?",
  "expected_plan": "NONE"
}

Ideally:

Question
   ↓
RAG
   ↓
No sufficiently relevant documents
   ↓
No recommendation

This is called abstention.


35. Why Abstention Matters

A bad AI system may behave like:

Unknown Question
      ↓
Find nearest random document
      ↓
Give confident answer

A safer design is:

Unknown Question
      ↓
Search knowledge
      ↓
Insufficient evidence
      ↓
Do not make recommendation
      ↓
Clarify or escalate

This is especially important in enterprise AI systems.


36. Complete RAG Lifecycle

Our RAG system now has two major phases.

Phase 1 — Ingestion

GlobalNet Plan JSON
        ↓
Document Loader
        ↓
Chunking
        ↓
Metadata
        ↓
Embedding Model
        ↓
Embeddings
        ↓
ChromaDB

This happens when knowledge is created or updated.

Phase 2 — Retrieval

Customer Question
        ↓
Query Embedding
        ↓
ChromaDB Search
        ↓
Top-K Chunks
        ↓
Distance Filtering
        ↓
Relevant Context
        ↓
Gemini
        ↓
Recommendation

37. Where the LLM Fits in RAG

A common misunderstanding is that the vector database generates the answer.

It does not.

The responsibilities are different.

Sentence Transformer
        ↓
Creates embeddings

ChromaDB
        ↓
Finds relevant knowledge

Gemini
        ↓
Reasons over retrieved knowledge
and generates the response

This separation is fundamental to understanding RAG.


38. What Happens When Plan Data Changes?

Suppose:

PREMIUM
500 Mbps
₹1099

changes to:

PREMIUM
600 Mbps
₹1199

We update:

globalnet_plans.json

Then re-run:

python scripts/ingest_plans.py

Because we use ChromaDB upsert, records with matching IDs are updated.

The knowledge flow becomes:

Updated Business Data
        ↓
Re-ingestion
        ↓
New Embeddings
        ↓
Updated Vector Records
        ↓
Future Queries
        ↓
Updated Knowledge

This is far better than manually changing AI prompts throughout the application.


39. Important Production Considerations

This implementation is still a learning architecture.

For a production system, I would add:

Document versioning
Incremental ingestion
Delete handling
Metadata filtering
Hybrid search
Reranking
Structured LLM output
Authentication
Authorization
Secrets management
Observability
Tracing
Prompt versioning
Evaluation pipelines
Error handling
Retries
Caching
Audit logs

The evaluation dataset would also need to be much larger and based on realistic customer queries.


40. Future RAG Architecture

The next evolution can look like:

                    Customer
                       |
                       v
                FastAPI / API Gateway
                       |
                       v
                Agent Orchestrator
                       |
          +------------+------------+
          |                         |
          v                         v
    Upsell Agent                RAG Agent
                                    |
                                    v
                             Query Rewriter
                                    |
                                    v
                              Vector Search
                                    |
                                    v
                            Metadata Filter
                                    |
                                    v
                                Reranker
                                    |
                                    v
                           Relevant Context
                                    |
                                    v
                              Plan Agent
                                    |
                                    v
                                  Gemini
                                    |
                                    v
                              Offer Agent
                                    |
                                    v
                             Final Response

41. What I Learned From This RAG Implementation

This project covers several important GenAI engineering concepts.

Knowledge Management

External knowledge
JSON documents
Document loading
Knowledge updates

Embeddings

Text → vectors
Semantic similarity
Embedding models

Vector Databases

ChromaDB
Persistent collections
Vector storage
Similarity search

Chunking

Document decomposition
Chunk IDs
Chunk types
Metadata

Retrieval

Semantic search
Top-K retrieval
Distance scores
Relevance thresholds

Generation

Retrieved context
Prompt grounding
Gemini
Agent reasoning

Evaluation

Evaluation datasets
Top-K retrieval
Top-1 accuracy
Failure analysis
Abstention

42. RAG vs Fine-Tuning

For this use case, changing plan information does not require retraining Gemini.

For example:

Old price → ₹1099
New price → ₹1199

With RAG:

Update Knowledge
      ↓
Re-index
      ↓
Retrieve New Information

This makes RAG particularly useful for frequently changing business knowledge.

Fine-tuning and RAG solve different problems.

RAG is generally appropriate when the model needs access to changing or private knowledge.

Fine-tuning is more appropriate when we want to alter model behavior, style, task patterns, or specialized capabilities.

They can also be combined.


43. Final Architecture So Far

At this stage, the GlobalNet project looks like:

                         Customer
                            |
                            v
                         FastAPI
                            |
                            v
                    +---------------+
                    | Upsell Agent  |
                    +-------+-------+
                            |
                            v
                         Gemini
                            |
                            v
                    +---------------+
                    |  Plan Agent   |
                    +-------+-------+
                            |
                            v
                       RAG Layer
                            |
                            v
                Sentence Transformer
                            |
                            v
                         ChromaDB
                            |
                            v
                   Relevant Chunks
                            |
                            v
                   Relevance Filter
                            |
                            v
                         Gemini
                            |
                            v
                  Plan Recommendation
                            |
                            v
                    +---------------+
                    |  Offer Agent  |
                    +-------+-------+
                            |
                            v
                         Gemini
                            |
                            v
                  Personalized Offer

And separately:

                    Evaluation Dataset
                            |
                            v
                    RAG Evaluation
                            |
               +------------+------------+
               |            |            |
               v            v            v
             Top-K        Top-1      Abstention

44. From Simple AI App to AI Engineering

The progression of this project so far has been:

Python
   ↓
FastAPI
   ↓
Vertex AI
   ↓
Gemini
   ↓
Prompt Engineering
   ↓
Upsell Agent
   ↓
Plan Agent
   ↓
Offer Agent
   ↓
External Knowledge
   ↓
Document Processing
   ↓
Chunking
   ↓
Embeddings
   ↓
ChromaDB
   ↓
Vector Search
   ↓
RAG
   ↓
Relevance Filtering
   ↓
RAG Evaluation

This progression is important because building a useful GenAI system requires more than simply calling an LLM API.

The surrounding engineering determines how knowledge is retrieved, controlled, evaluated and delivered.


Conclusion

In this part of the GlobalNet Support Agent project, I transformed the Plan Agent from a hardcoded prompt-based implementation into a RAG-powered architecture.

We implemented:

GlobalNet Knowledge Base
        ↓
Document Loader
        ↓
Chunking
        ↓
Metadata
        ↓
Sentence Transformer
        ↓
Embeddings
        ↓
ChromaDB
        ↓
Semantic Search
        ↓
Top-K Retrieval
        ↓
Distance Filtering
        ↓
Plan Agent
        ↓
Gemini

We also added an evaluation layer:

Evaluation Questions
        ↓
Expected Plans
        ↓
RAG Retrieval
        ↓
Compare Results
        ↓
Top-K / Top-1 Accuracy

The next stage is to improve the retrieval layer further by returning structured RAG results, adding metadata filtering, improving retrieval quality, and eventually introducing techniques such as reranking and hybrid search.

The long-term objective is to evolve this project into a more complete AI support platform:

Multi-Agent AI
      +
RAG
      +
Customer Database
      +
Firestore
      +
Cloud Run
      +
Observability
      +
Automated Evaluation
      =
Production-Style AI Support Platform

Tuesday, 6 October 2026

Part 1 — Building a Production-Style RAG Pipeline for a Telecom AI Support Agent

In this tutorial, I am building a practical AI-powered Telecom Customer Support Agent using Python, FastAPI, Google Cloud, Vertex AI and Gemini.

The goal is to build an AI system that can understand a telecom customer's problem and then:

  1. Analyze whether there is a legitimate upsell opportunity.

  2. Recommend the most appropriate telecom plan.

  3. Generate a personalized customer offer.

The project is called:

GlobalNet Support Agent




This is not just a chatbot. The objective is to build a modular AI-agent architecture that can later be extended with:

  • RAG

  • Firestore

  • Customer data

  • Plan knowledge base

  • Email notifications

  • Cloud Run

  • Cloud Scheduler

  • Monitoring

  • Multiple AI agents

  • Kubernetes integration

This article covers the implementation completed so far.


1. Project Architecture

The current architecture is:

                         Customer
                            |
                            v
                    FastAPI /support
                            |
                            v
                    +---------------+
                    |  Upsell Agent |
                    +---------------+
                            |
                            v
                       Gemini AI
                            |
                            v
                    +---------------+
                    |   Plan Agent  |
                    +---------------+
                            |
                            v
                       Gemini AI
                            |
                            v
                    +---------------+
                    |  Offer Agent  |
                    +---------------+
                            |
                            v
                    Personalized Offer

The current technology stack is:

Python
FastAPI
Pydantic
Google Gen AI SDK
Google Cloud Vertex AI
Gemini 2.5 Flash
PowerShell
VS Code

2. Final Directory Structure So Far

The project currently has the following structure:

GlobalNet-Support-Agent-GCP/
│
├── app/
│   ├── __init__.py
│   │
│   ├── main.py
│   │
│   ├── agents/
│   │   ├── __init__.py
│   │   ├── upsell_agent.py
│   │   ├── plan_agent.py
│   │   └── offer_agent.py
│   │
│   ├── services/
│   │   ├── __init__.py
│   │   ├── vertex_ai.py
│   │   ├── rag.py
│   │   ├── firestore.py
│   │   └── email.py
│   │
│   └── models/
│       └── customer.py
│
├── jobs/
│   └── scan_inactive_customers.py
│
├── data/
│   └── plans/
│
├── test_vertex.py
│
├── requirements.txt
├── Dockerfile
├── .dockerignore
└── README.md

Some directories such as rag.py, firestore.py, email.py, and scheduled jobs are part of the planned architecture and will be implemented in later parts.


3. Create the GCP Project

The Google Cloud project used for this application is:

globalnet-support-agent

Set the project using the Google Cloud CLI:

gcloud config set project globalnet-support-agent

Expected output:

Updated property [core/project].

4. Enable Required Google Cloud APIs

The following APIs are required for the project:

gcloud services enable `
run.googleapis.com `
cloudbuild.googleapis.com `
artifactregistry.googleapis.com `
aiplatform.googleapis.com `
firestore.googleapis.com `
storage.googleapis.com `
cloudscheduler.googleapis.com `
secretmanager.googleapis.com `
logging.googleapis.com

These services will eventually provide:

Vertex AI
Cloud Run
Cloud Build
Artifact Registry
Firestore
Cloud Storage
Cloud Scheduler
Secret Manager
Cloud Logging

5. Configure Google Cloud Authentication

For local development, Application Default Credentials can be configured using:

gcloud auth application-default login

The project can also be configured as the quota project:

gcloud auth application-default set-quota-project globalnet-support-agent

6. Configure Environment Variables

The application uses the following environment variables.

In PowerShell:

$env:GOOGLE_CLOUD_PROJECT="globalnet-support-agent"
$env:GOOGLE_CLOUD_LOCATION="asia-south1"

These values are used by the Vertex AI service.


7. Vertex AI Service

Instead of putting Gemini code directly inside every agent, I created a reusable service.

File:

app/services/vertex_ai.py

Code:

from google import genai
from google.genai.types import GenerateContentConfig
import os


PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
LOCATION = os.environ.get(
    "GOOGLE_CLOUD_LOCATION",
    "asia-south1"
)


client = genai.Client(
    vertexai=True,
    project=PROJECT_ID,
    location=LOCATION,
)


def generate_text(prompt: str) -> str:

    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
        config=GenerateContentConfig(
            temperature=0.2,
        ),
    )

    return response.text

This creates one reusable function:

generate_text()

All AI agents can use this function.


8. Test Gemini Independently

Before connecting Gemini to FastAPI, I created:

test_vertex.py

Code:

from app.services.vertex_ai import generate_text


prompt = """
You are a telecom customer support AI.

A customer says:

"My internet is very slow and I need more bandwidth."

Explain whether there could be an upsell opportunity.
"""


print("Calling Gemini...")


result = generate_text(prompt)


print("\n===== GEMINI RESPONSE =====\n")
print(result)

Run:

python test_vertex.py

If everything is configured correctly, Gemini returns a response.

This is an important development practice:

First verify the AI service independently, then integrate it with the application.


9. FastAPI Application

The main API is:

app/main.py

FastAPI is used to expose the AI agent through an HTTP API.

The application starts with:

from fastapi import FastAPI
from pydantic import BaseModel

The FastAPI application is created using:

app = FastAPI(
    title="GlobalNet Support Agent",
    description="AI-powered telecom customer support agent",
    version="1.0.0",
)

10. Customer Request Model

The /support API receives customer information.

class SupportRequest(BaseModel):

    customer_id: str

    chat_text: str

    loyalty_status: str

    current_plan: str

    current_plan_desc: str

    tenure_months: int

    customer_email: str

    customer_name: str

    call_type: str

Pydantic automatically validates this request.


11. Health Check API

The application contains a basic health endpoint:

@app.get("/health")
def health():

    return {
        "status": "healthy"
    }

Testing:

GET /health

Expected response:

{
  "status": "healthy"
}

This endpoint will become useful later for Cloud Run and production monitoring.


12. Root Endpoint

The root endpoint is:

@app.get("/")
def root():

    return {
        "status": "ok",
        "service": "GlobalNet Support Agent",
        "message": "API is running",
    }

13. Upsell Agent

The first AI agent is the:

Upsell Agent

File:

app/agents/upsell_agent.py

Its responsibility is to determine whether the customer's problem represents a genuine upgrade opportunity.

Complete code:

from app.services.vertex_ai import generate_text


def analyze_upsell(customer) -> str:

    prompt = f"""
You are a telecom customer support upsell AI.

Analyze the customer information below.

Customer Information:
- Customer ID: {customer.customer_id}
- Customer Name: {customer.customer_name}
- Loyalty Status: {customer.loyalty_status}
- Current Plan: {customer.current_plan}
- Current Plan Description: {customer.current_plan_desc}
- Tenure: {customer.tenure_months} months
- Call Type: {customer.call_type}

Customer Message:
{customer.chat_text}

Determine:

1. What is the customer's main problem?
2. Is there a legitimate upsell opportunity?
3. Why is there or isn't there an opportunity?
4. What type of plan would be appropriate?
5. What should the support agent do next?

Important:

- Do not recommend an upgrade just to increase revenue.
- First determine whether the customer's problem could be
  caused by a technical issue.
- If a technical issue is likely, recommend troubleshooting first.
- Only recommend an upsell when the customer's actual usage
  justifies a higher plan.

Provide a clear and concise answer.
"""

    return generate_text(prompt)

14. Why the Upsell Agent Checks Technical Problems

This is important in a real telecom environment.

A customer saying:

"My internet is very slow."

does not automatically mean:

Customer needs a bigger plan.

The problem could be:

Router problem
       |
Wi-Fi interference
       |
Packet loss
       |
DNS problem
       |
ISP network problem
       |
Congestion
       |
Signal issue
       |
Only then → insufficient bandwidth

Therefore, the AI should not automatically try to sell something.

The prompt explicitly tells the AI:

Do not recommend an upgrade just to increase revenue.

This makes the agent more customer-centric.


15. Plan Agent

The second agent is:

Plan Agent

File:

app/agents/plan_agent.py

Its responsibility is to select the most appropriate plan.

Complete code:

from app.services.vertex_ai import generate_text


def recommend_plan(customer, upsell_analysis: str) -> str:

    prompt = f"""
You are a telecom plan recommendation AI.

Analyze the customer information and upsell analysis below.

Customer Information:
- Customer ID: {customer.customer_id}
- Customer Name: {customer.customer_name}
- Current Plan: {customer.current_plan}
- Current Plan Description: {customer.current_plan_desc}
- Loyalty Status: {customer.loyalty_status}
- Tenure: {customer.tenure_months} months
- Call Type: {customer.call_type}

Customer Message:
{customer.chat_text}

Upsell Analysis:
{upsell_analysis}

Available Example Plans:

1. BASIC
   Speed: 100 Mbps

2. STANDARD
   Speed: 300 Mbps

3. PREMIUM
   Speed: 500 Mbps

4. ULTRA
   Speed: 1 Gbps

Rules:

1. Do not recommend an upgrade just to increase revenue.
2. If the problem appears to be a technical issue, recommend
   troubleshooting before upgrading.
3. If the customer genuinely needs more bandwidth, recommend
   the lowest plan that can satisfy the requirement.
4. Consider the customer's current plan.
5. Consider the customer's stated usage.
6. Explain why the recommended plan is appropriate.
7. Mention the current plan.
8. Mention the recommended plan.
9. Keep the response concise and suitable for a support agent.

Return:

Current Plan:
Recommended Plan:
Reason:
Next Action:
"""

    return generate_text(prompt)

16. Example Plan Catalog

Currently the plans are hardcoded for learning purposes:

BASIC
100 Mbps

STANDARD
300 Mbps

PREMIUM
500 Mbps

ULTRA
1 Gbps

Later, these values will not be hardcoded.

They will come from a real knowledge source:

Plan Agent
     |
     v
RAG
     |
     v
Plan Knowledge Base
     |
     v
GlobalNet Plans

This will be implemented in a future part.


17. Offer Agent

The third agent is:

Offer Agent

File:

app/agents/offer_agent.py

Its responsibility is to create a personalized offer.

Complete code:

from app.services.vertex_ai import generate_text


def generate_offer(customer, plan_recommendation: str) -> str:

    prompt = f"""
You are a telecom customer retention and offer AI.

Analyze the customer information below and create an appropriate
personalized telecom offer.

Customer Information:
- Customer ID: {customer.customer_id}
- Customer Name: {customer.customer_name}
- Current Plan: {customer.current_plan}
- Current Plan Description: {customer.current_plan_desc}
- Loyalty Status: {customer.loyalty_status}
- Tenure: {customer.tenure_months} months
- Call Type: {customer.call_type}

Customer Message:
{customer.chat_text}

Plan Recommendation:
{plan_recommendation}

Offer Rules:

1. Do not create an offer if there is no legitimate upgrade
   opportunity.
2. If a technical issue is likely, recommend troubleshooting
   instead of an upgrade offer.
3. GOLD or higher loyalty customers may receive a stronger
   retention benefit.
4. Long-tenure customers may receive a loyalty benefit.
5. Do not invent unrealistic discounts.
6. The offer should be simple and easy for a support agent
   to explain to the customer.
7. Explain why the customer is receiving the offer.
8. The offer should match the recommended plan.

Return the result in this format:

Offer Decision:
Recommended Plan:
Monthly Benefit:
Loyalty Benefit:
Reason:
Agent Action:
Customer Message:
"""

    return generate_text(prompt)

18. Complete main.py

The current complete main.py is:

from fastapi import FastAPI
from pydantic import BaseModel

from app.agents.upsell_agent import analyze_upsell
from app.agents.plan_agent import recommend_plan
from app.agents.offer_agent import generate_offer


app = FastAPI(
    title="GlobalNet Support Agent",
    description="AI-powered telecom customer support agent",
    version="1.0.0",
)


class SupportRequest(BaseModel):

    customer_id: str
    chat_text: str
    loyalty_status: str
    current_plan: str
    current_plan_desc: str
    tenure_months: int
    customer_email: str
    customer_name: str
    call_type: str


@app.get("/")
def root():

    return {
        "status": "ok",
        "service": "GlobalNet Support Agent",
        "message": "API is running",
    }


@app.get("/health")
def health():

    return {
        "status": "healthy"
    }


@app.post("/support")
def support(request: SupportRequest):

    # Step 1: Analyze upsell opportunity
    upsell_result = analyze_upsell(request)

    # Step 2: Recommend appropriate plan
    plan_result = recommend_plan(
        request,
        upsell_result
    )

    # Step 3: Generate personalized offer
    offer_result = generate_offer(
        request,
        plan_result
    )

    return {
        "status": "success",
        "customer_id": request.customer_id,
        "customer_name": request.customer_name,
        "upsell_analysis": upsell_result,
        "plan_recommendation": plan_result,
        "personalized_offer": offer_result,
    }

19. Start the Application

From the project root:

cd "C:\Users\Raj Kumar Gupta\Desktop\Raj\minikube\GlobalNet-Support-Agent-GCP"

Set the environment variables:

$env:GOOGLE_CLOUD_PROJECT="globalnet-support-agent"
$env:GOOGLE_CLOUD_LOCATION="asia-south1"

Start FastAPI:

uvicorn app.main:app --reload

Expected:

Uvicorn running on http://127.0.0.1:8000

20. Open Swagger

FastAPI automatically provides Swagger UI.

Open:

http://127.0.0.1:8000/docs

You should see:

GET  /
GET  /health
POST /support

21. Test /support

Select:

POST /support

Click:

Try it out

Use this request:

{
  "customer_id": "1001",
  "chat_text": "My internet is very slow and I need much higher bandwidth because I have many devices and use 4K streaming.",
  "loyalty_status": "GOLD",
  "current_plan": "BASIC",
  "current_plan_desc": "100 Mbps",
  "tenure_months": 48,
  "customer_email": "your-email@example.com",
  "customer_name": "Raj",
  "call_type": "online"
}

Click:

Execute

22. What Happens Internally?

The request enters:

POST /support

Then:

SupportRequest
       |
       v
analyze_upsell()
       |
       v
Gemini
       |
       v
upsell_result

Then:

upsell_result
       |
       v
recommend_plan()
       |
       v
Gemini
       |
       v
plan_result

Then:

plan_result
       |
       v
generate_offer()
       |
       v
Gemini
       |
       v
offer_result

Finally:

JSON Response

23. Final Response Structure

The API returns a structure similar to:

{
  "status": "success",
  "customer_id": "1001",
  "customer_name": "Raj",
  "upsell_analysis": "...",
  "plan_recommendation": "...",
  "personalized_offer": "..."
}

The exact AI text will vary because Gemini generates the response dynamically.


24. Complete Request Flow

The complete flow is now:

                         +----------------+
                         |    Customer    |
                         +-------+--------+
                                 |
                                 v
                         +---------------+
                         |    FastAPI    |
                         |   /support    |
                         +-------+-------+
                                 |
                                 v
                     +-----------------------+
                     |    Upsell Agent       |
                     |                       |
                     | Is upgrade justified? |
                     +----------+------------+
                                |
                                v
                           +---------+
                           | Gemini  |
                           +----+----+
                                |
                                v
                     +-----------------------+
                     |      Plan Agent       |
                     |                       |
                     | Which plan is best?   |
                     +----------+------------+
                                |
                                v
                           +---------+
                           | Gemini  |
                           +----+----+
                                |
                                v
                     +-----------------------+
                     |     Offer Agent       |
                     |                       |
                     | What offer should     |
                     | we provide?           |
                     +----------+------------+
                                |
                                v
                           +---------+
                           | Gemini  |
                           +----+----+
                                |
                                v
                     +-----------------------+
                     |   Final API Response  |
                     +-----------------------+

25. Why Use Multiple Agents?

Instead of creating one large prompt, the application separates responsibilities.

Upsell Agent

Answers:

Should we upgrade the customer?

Plan Agent

Answers:

Which plan is appropriate?

Offer Agent

Answers:

What personalized offer should we provide?

This separation makes the application easier to:

  • Maintain

  • Test

  • Debug

  • Extend

  • Replace individual agents

  • Add business rules

  • Add RAG

  • Add databases


26. Current Limitations

The current implementation is a learning prototype.

There are several things that need to be improved before production.

Hardcoded plans

Currently:

BASIC → 100 Mbps
STANDARD → 300 Mbps
PREMIUM → 500 Mbps
ULTRA → 1 Gbps

These should come from a database or knowledge base.

Free-form AI responses

The agents currently return strings.

A production implementation should return structured JSON.

For example:

{
  "upgrade_required": true,
  "recommended_plan": "PREMIUM",
  "reason": "Customer has multiple devices and 4K streaming requirements",
  "next_action": "Offer plan upgrade"
}

No RAG yet

The Plan Agent does not yet retrieve information from a real GlobalNet knowledge base.

No Firestore yet

Customer information is currently supplied directly through the API request.

No authentication

The API currently has no authentication or authorization.

No production monitoring

Cloud Logging, metrics and tracing still need to be integrated.


27. What We Will Build Next

The next stage will make the project significantly more realistic.

The planned architecture is:

Customer
   |
   v
FastAPI
   |
   v
Agent Orchestrator
   |
   +------------------+
   |                  |
   v                  v
Upsell Agent       RAG System
   |                  |
   |                  v
   |            Plan Knowledge
   |                  |
   +--------+---------+
            |
            v
       Plan Agent
            |
            v
       Offer Agent
            |
            v
         Firestore
            |
            v
        Email Agent
            |
            v
      Customer Notification

Future components will include:

RAG
ChromaDB / Vector Database
Embeddings
Firestore
Customer Database
Plan Knowledge Base
Email Service
Cloud Run
Artifact Registry
Cloud Scheduler
Cloud Logging

28. Learning Lessons From This Project

This project demonstrates several important Python and AI engineering concepts.

Python

We are learning:

Functions
Imports
Modules
Packages
Classes
Type validation
Virtual environments
Environment variables
Exception handling

FastAPI

We are learning:

REST API
Endpoints
POST requests
Pydantic models
Swagger
Health checks
API integration

Generative AI

We are learning:

LLM integration
Prompt engineering
Agent design
Multi-agent architecture
Context passing
AI decision making

Cloud

We are learning:

Google Cloud
Vertex AI
Cloud Run
Artifact Registry
Firestore
Cloud Scheduler
Secret Manager
Cloud Logging

29. Conclusion

We have now created the first working version of the GlobalNet Support Agent.

The application can:

Receive customer request
        ↓
Analyze upsell opportunity
        ↓
Recommend a suitable plan
        ↓
Generate a personalized offer
        ↓
Return the result through REST API

The most important design principle is that the AI should not simply try to sell a more expensive plan.

The system should first understand the customer's actual problem.

For example:

Customer:
"My internet is slow."

          ↓

AI investigates

          ↓

Could be technical issue?
          |
       YES ─────→ Troubleshoot first
          |
         NO
          |
          v
Does usage justify more bandwidth?
          |
       YES
          |
          v
Recommend appropriate plan
          |
          v
Generate personalized offer


Wednesday, 2 September 2026

Container OOM vs Node OOM in Kubernetes

Memory-related incidents are among the most common and confusing problems in Kubernetes production environments.

A Pod restarts unexpectedly, an application becomes unavailable, or a Kubernetes node suddenly becomes unstable. One of the first things an SRE may notice is:

OOMKilled

But does OOMKilled always mean that the Kubernetes node ran out of memory?

No.

Understanding the difference between Container OOM, Node Memory Pressure, Pod Eviction, and Node-level OOM is extremely important for Kubernetes SREs.


1. Container OOM vs Node OOM

The simplest way to understand the difference is:

Container OOM is a container-level memory problem. Node OOM is a node-level memory problem.

AreaContainer OOMNode OOM
ScopeOne container/processEntire Kubernetes node
Main causeContainer exceeds its memory limitNode runs critically low on memory
Typical evidenceOOMKilledMemoryPressure, kernel OOM messages
Who gets killed?Usually the offending container/processLinux kernel may kill processes/containers
Pod behaviorContainer may restartPods may be evicted, killed, or become unstable
Other workloads affectedUsually limitedPotentially many workloads
Memory limit involved?UsuallyNot necessarily
SeverityApplication/container-levelNode/cluster-level

The distinction becomes clearer with a production example.


2. What is Container OOM?

Consider a Kubernetes node with 64 GB of memory.

A Pod is running an application with the following configuration:

resources:
  requests:
    memory: "1Gi"
  limits:
    memory: "2Gi"

The container's application gradually consumes more memory:

500 MB
   ↓
1 GB
   ↓
1.5 GB
   ↓
2 GB
   ↓
2.2 GB

The container has a memory limit of 2 GiB.

If the container attempts to consume memory beyond its allowed limit, the process can be killed due to an out-of-memory condition.

Kubernetes may then report:

Reason: OOMKilled

You can investigate the Pod with:

kubectl describe pod myapp-xxx -n production

You may see:

Last State:
  Terminated:
    Reason: OOMKilled
    Exit Code: 137

This is an important clue that the container experienced a memory-related termination.


3. What does Exit Code 137 mean?

You will frequently encounter:

Exit Code: 137

The number comes from:

128 + 9 = 137

Signal 9 is:

SIGKILL

Therefore, exit code 137 generally indicates that the process was killed with SIGKILL.

However, an important SRE rule is:

Do not assume that every exit code 137 automatically means OOM.

Always verify the termination reason and supporting system evidence.

For example:

kubectl describe pod <pod> -n <namespace>

Look for:

Reason: OOMKilled

Then check the node and kernel logs if necessary.


4. What is Node OOM?

Now consider a different situation.

Suppose a Kubernetes node has 64 GB of memory:

Kubernetes Node
Memory = 64 GB

Several workloads are consuming memory:

Pod A       → 15 GB
Pod B       → 12 GB
Pod C       → 18 GB
Pod D       → 10 GB
System      →  8 GB
---------------------
Total       → 63 GB

The node is now under severe memory pressure.

If memory consumption continues and the Linux kernel cannot satisfy memory allocations, the kernel may invoke the OOM killer.

This is fundamentally different from a single container exceeding its configured memory limit.

The problem is now:

The node itself is running out of available memory.


5. How to identify Node Memory Pressure

From the Kubernetes control plane, check the node:

kubectl describe node <node-name>

Look at the Conditions section.

You may find:

Conditions:
  MemoryPressure   True

This means the kubelet has detected memory pressure on the node.

You should then investigate the actual node.

Run:

free -h

Example:

               total        used        free
Mem:             64Gi         62Gi       500Mi
Swap:               0           0           0

Other useful commands include:

top
vmstat 1
ps aux --sort=-%mem

These help identify which processes are consuming memory.


6. Check Linux Kernel Logs

For a node-level OOM investigation, kernel logs are extremely important.

Run:

dmesg -T | grep -i -E "oom|out of memory|killed process"

Or:

journalctl -k | grep -i -E "oom|out of memory|killed process"

You may find something similar to:

Out of memory: Killed process 12345 (java)

This is strong evidence that the Linux kernel invoked the OOM killer.

At this point, your investigation has moved beyond Kubernetes objects and into the Linux operating system layer.

That is an important skill for a Kubernetes SRE.


7. The SRE Mental Model

Think about a Kubernetes node like this:

                 KUBERNETES NODE
        ┌─────────────────────────────┐
        │                             │
        │  Pod A                      │
        │  ┌───────────────────────┐  │
        │  │ Container             │  │
        │  │ Memory limit: 2 GiB   │  │
        │  └───────────────────────┘  │
        │                             │
        │  Pod B                      │
        │                             │
        │  Pod C                      │
        │                             │
        │  kubelet                    │
        │  system processes           │
        │                             │
        └─────────────────────────────┘

Container OOM

Container memory usage
        ↓
Exceeds container limit
        ↓
Container/process killed
        ↓
Kubernetes reports OOMKilled
        ↓
Container may restart

Node-level OOM

Total node memory consumption
        ↓
Available memory becomes critically low
        ↓
Linux kernel cannot satisfy allocation
        ↓
Kernel OOM killer
        ↓
Process/container killed
        ↓
Potential impact to multiple workloads

8. Don't confuse OOM with Pod Eviction

This is where Kubernetes troubleshooting becomes more interesting.

There are at least three different situations an SRE should distinguish.

Situation 1 — Container OOM

Container exceeds memory limit
        ↓
Container/process killed
        ↓
OOMKilled

Situation 2 — Node Memory Pressure

Node available memory becomes low
        ↓
Kubelet detects memory pressure
        ↓
MemoryPressure = True
        ↓
Kubernetes may evict Pods

Situation 3 — Node-level Linux OOM

Node cannot satisfy memory allocation
        ↓
Linux kernel OOM killer
        ↓
Process/container killed

These situations are related, but they are not the same event.


9. Production Troubleshooting Methodology

When investigating a memory-related incident, don't immediately jump to:

"The application has an OOM."

Follow an evidence-based approach.

Step 1 — Check Pod status

kubectl get pod <pod> -n <namespace>

Example:

NAME        READY   STATUS             RESTARTS
myapp-01    0/1     CrashLoopBackOff   8

Step 2 — Describe the Pod

kubectl describe pod <pod> -n <namespace>

Look for:

Last State:
  Terminated:
    Reason: OOMKilled
    Exit Code: 137

Also check Events.


Step 3 — Check container logs

kubectl logs <pod> -n <namespace>

If the container has restarted:

kubectl logs <pod> -n <namespace> --previous

The --previous option is particularly useful because the current container may have already restarted.


10. Check Resource Requests and Limits

Inspect the Pod configuration:

kubectl get pod <pod> -n <namespace> -o yaml

Look for:

resources:
  requests:
    memory: "1Gi"
  limits:
    memory: "2Gi"

Ask:

  • Is the memory limit too low?

  • Is the application memory usage increasing?

  • Is there a memory leak?

  • Are requests and limits configured correctly?

  • Has application behavior changed recently?


11. Check the Node

Find which node is running the Pod:

kubectl get pod <pod> -n <namespace> -o wide

Then:

kubectl describe node <node-name>

Check:

MemoryPressure
Allocatable memory
Allocated resources
Conditions
Events

12. Check Node Memory from Linux

On the affected node:

free -h
top
vmstat 1
ps aux --sort=-%mem

Also check:

df -h

Although df -h primarily checks filesystem capacity rather than RAM, it is useful during broader node-health investigations because memory incidents can occur alongside other resource exhaustion problems.


13. Check Kernel Evidence

Finally:

dmesg -T | grep -i -E "oom|out of memory|killed process"

or:

journalctl -k | grep -i -E "oom|out of memory|killed process"

Now correlate:

Kubernetes evidence
        +
Container evidence
        +
Node evidence
        +
Linux kernel evidence
        ↓
Root Cause

This is the approach an experienced SRE should follow.


14. Example Production RCA

Suppose an application Pod restarted several times.

You run:

kubectl describe pod payment-01 -n production

and find:

Last State:
  Terminated:
    Reason: OOMKilled
    Exit Code: 137

The Pod configuration shows:

resources:
  requests:
    memory: "1Gi"
  limits:
    memory: "2Gi"

Monitoring shows the application memory usage increased steadily:

10:00 → 1.1 GiB
10:10 → 1.4 GiB
10:20 → 1.7 GiB
10:30 → 1.9 GiB
10:35 → 2.0 GiB
10:36 → OOMKilled

The evidence indicates:

Application memory consumption increased
             ↓
Container reached 2 GiB limit
             ↓
Container was killed
             ↓
Kubernetes reported OOMKilled
             ↓
Container restarted

This is primarily a container-level OOM, not proof of a node-level OOM.


15. Example Node-Level Incident

Now consider:

Node memory = 64 GiB

Pod A = 15 GiB
Pod B = 12 GiB
Pod C = 18 GiB
Pod D = 10 GiB
System = 8 GiB

The node becomes heavily memory constrained.

You observe:

kubectl describe node worker-01
MemoryPressure: True

Then:

journalctl -k | grep -i oom

returns:

Out of memory: Killed process 12345 (java)

Now you have evidence of a node-level memory exhaustion event.

The impact can potentially extend beyond one application.


16. Interview Question

A common senior Kubernetes SRE interview question is:

A Pod shows OOMKilled. Does that mean the Kubernetes node experienced an OOM?

Correct answer

No.

OOMKilled indicates that the container/process was killed because of an out-of-memory condition, commonly because the container exceeded its configured memory limit.

It does not by itself prove that the Kubernetes node experienced a system-wide OOM.

To determine whether the node itself experienced memory exhaustion, check:

kubectl describe node <node>

Look for:

MemoryPressure: True

Then check the node's Linux/kernel evidence:

free -h
dmesg -T | grep -i oom
journalctl -k | grep -i oom

This distinction is extremely important in production troubleshooting.


17. Senior SRE Interview Answer

If an interviewer asks:

"Explain the difference between Container OOM and Node OOM."

A strong answer would be:

Container OOM is a container-level memory exhaustion condition, typically occurring when a container exceeds its configured memory limit. Kubernetes may report the container termination as OOMKilled, and the container may subsequently restart.

Node OOM is a node-level memory exhaustion condition where the overall system runs critically low on memory. The Linux kernel may invoke the OOM killer and terminate processes, potentially affecting multiple workloads. Kubernetes may also detect node memory pressure and evict Pods before the system reaches a hard kernel OOM condition.

During troubleshooting, I would correlate Pod termination reasons, resource requests and limits, node MemoryPressure, kubelet events, node memory utilization, and Linux kernel logs to determine the actual root cause.

That demonstrates production troubleshooting rather than simple Kubernetes theory.


18. Quick SRE Cheat Sheet

                 MEMORY INCIDENT
                       │
                       ▼
               Check Pod status
                       │
                       ▼
              kubectl describe pod
                       │
              ┌────────┴────────┐
              ▼                 ▼
          OOMKilled?          Other?
              │                 │
              ▼                 ▼
       Check limits          Continue
       Check usage           troubleshooting
       Check logs
              │
              ▼
       Check affected node
              │
              ▼
     MemoryPressure = True?
              │
        ┌─────┴─────┐
        ▼           ▼
       YES          NO
        │            │
        ▼            ▼
 Check node       Investigate
 memory           container/app
        │
        ▼
 Check kernel logs
        │
        ▼
 Root Cause

Conclusion

The most important lesson is:

OOMKilled does not automatically mean "the Kubernetes node ran out of memory."

Always determine the scope of the memory problem:

Container
   ↓
Pod
   ↓
Node
   ↓
Cluster

For a Kubernetes SRE, the goal is not simply to identify that "OOM happened."

The real goal is to answer:

What ran out of memory? Why did it happen? What evidence proves it? What was the impact? How do we fix it? And how do we prevent it from happening again?