DEV Community: Mainak Bhattacharjee

Context-Aware ServiceNow: Feeding the Right Data at the Right Time with RAG

Mainak Bhattacharjee — Wed, 10 Jun 2026 20:32:03 +0000

Recently, I’ve been diving deep into Retrieval-Augmented Generation (RAG) in Python, and the potential is incredible. The idea that we can feed exact, contextual information to an LLM so it can generate precise answers—without burning tokens on 1,000-page documents or forcing users to dig through endless search results—is a game-changer.

Naturally, my journey started by building a prototype using popular abstractions like LangChain and ChromaDB. It worked, but it left me wondering: What is actually happening under the hood? High-level frameworks are great for speed, but they often mask low-level complexities and leave your architecture tightly coupled to a specific ecosystem. To truly understand the mechanics, I decided to ditch LangChain and ChromaDB altogether. I went back to the basics: handling API calls manually, managing data storage in a standard database, and writing my own retrieval queries.

And guess what? It worked.

Stripping away the magic gave me a much deeper appreciation for RAG architecture, and it proved you don't need a bloated tech stack to build something highly effective. With a solid grasp of the core mechanics, I decided to take this lightweight, low-level approach and port it directly into the ServiceNow ecosystem. Let's jump into how it works.

General Overview:
Let’s break the architecture down into its core components:

Data Ingestion: Extract your raw data as plain text (e.g., parsing PDFs).
Text Chunking: Divide the text into smaller, digestible segments to maintain context limits.
Vector Embeddings: Convert each text chunk into a high-dimensional vector array (represented as 32-bit floating-point numbers).
Storage: Store both the raw text chunks and their corresponding floating-point arrays into a standard database.
In Python we will use sqlite, in ServiceNow we will use two custom tables: u_embeddings and u_hnsw_relations.
Custom Indexing: Since we aren't using a dedicated vector database, we need to build a multi-layered graph (like an HNSW index) from scratch for optimized, fast-vector searching.
Similarity Search: Compute the vector distance between the user’s query embedding and our stored chunks to extract the top-K most relevant results.
LLM Generation: Pass those top-K chunks as context, along with the system prompt and the user’s original query, to the LLM to synthesize the final answer.

Looking at it all laid out, it sounds like an uphill battle—and frankly, it is. But building it from scratch is where the real fun begins. Let’s dive into the code.

Data Ingestion:
The first phase of the pipeline is Data Ingestion. While my Python prototype relied on libraries like pypdf to parse static PDF files, the ServiceNow architecture shifts toward dynamic, platform-native data sources. Instead of PDFs, we will ingest data directly from Knowledge Articles (kb_knowledge), utilizing standard GlideRecord queries to programmatically extract the raw text fields.

Python:

reader = pypdf.PdfReader(file_path)
text = ""
for page in reader.pages:
    page_text = page.extract_text()
    if page_text:
        text += page_text + "\n"
return text

ServiceNow:

var kbGr = new GlideRecord("kb_knowledge");
kbGr.addEncodedQuery("workflow_state=published");
kbGr.query();
while (kbGr.next()) {
    var text = kbGr.getDisplayValue('text');
}

Now that we’ve extracted the raw text, the next phase is Text Chunking. We will implement a word-based chunking strategy with an overlap to ensure the LLM retains semantic context across boundaries.

To see how this works, let's look at a quick example. Suppose we have the text:
"ServiceNow is the best automation tool in the world"
If we set our chunk size to 5 and our overlap to 2, our script will slice the string into an array where each element contains exactly five words, and the last two words of any given chunk become the first two words of the next.

Visually, that sliding window looks like this:

[
"ServiceNow is the best automation",
"best automation tool in the",
"in the world"
]

Python:

words = re.findall(r"\b[\w'-]+\b", text)
text_len = len(words)
start = 0
chunk_list = []
while start < text_len:
    end = min(start + chunk_size, text_len)
    chunk = " ".join(words[start:end])
    chunk_list.append(chunk)
    start = start + chunk_size - overlap
return chunk_list

ServiceNow:

if (!text) return [];
var words = text.match(/\b[\w'-]+\b/g) || [];
var textLen = words.length;
var start = 0;
var chunkList = [];

while (start < textLen) {
    var end = Math.min(start + chunkSize, textLen);
    var chunk = words.slice(start, end).join(' ');
    chunkList.push(chunk);
    start = start + chunkSize - overlap;
    if (chunkSize <= overlap) {
        break;
    }
}
return chunkList;

With our chunks successfully generated, the next step is to transform that text into mathematical representations via Vector Embeddings. For this architecture, we will utilize HuggingFace's all-MiniLM-L6-v2 text embedding model.

The core mechanic here is straightforward: we pass our list of string chunks into the embedding model. In return, the model maps the semantic meaning of each chunk into a high-dimensional vector space. The response is a 2D array of float32 values, where each row represents a single text chunk transformed into a dense 384-dimensional vector embedding (an array of float32 numbers).

Python:

embed_api_url = f"https://router.huggingface.co/hf-inference/models/sentence-transformers/all-MiniLM-L6-v2/pipeline/feature-extraction"
headers = {"Authorization": f"Bearer {os.environ['HF_TOKEN']}"}
response = requests.post(
    embed_api_url,
    json={"inputs": {"sentences": chunk_list}, "options":{"wait_for_model": True}},
        headers=headers,
)
if response.status_code != 200:
    raise Exception(f"HF API Error: {response.status_code} - {response.text}")

return response.json()

ServiceNow:

var provider = new sn_cc.StandardCredentialsProvider();
var credential = provider.getCredentialByAliasID("fa0f862d93950710c987fc532bba1010");

var token = credential.getAttribute('api_key');

if (!token) {
    throw new Error("Unable to retrieve Hugging Face token from connection alias 'HF_API_TOKEN'.");
}

var endpoint = "https://router.huggingface.co/hf-inference/models/sentence-transformers/all-MiniLM-L6-v2/pipeline/feature-extraction";

var request = new sn_ws.RESTMessageV2();
request.setEndpoint(endpoint);
request.setHttpMethod('POST');

request.setRequestHeader('Authorization', 'Bearer ' + token);
request.setRequestHeader('Content-Type', 'application/json');

var payload = {
    "inputs": {
        "sentences": chunkList
    },
    "options": {
        "wait_for_model": true
    }
};
request.setRequestBody(JSON.stringify(payload));

var response = request.execute();
var statusCode = response.getStatusCode();
var responseBody = response.getBody();

token = null;

if (statusCode !== 200) {
    throw new Error("HF API Error: " + statusCode + " - " + responseBody);
}

return JSON.parse(responseBody);

At this point, we have all the pieces for our ingestion pipeline. But there’s a massive elephant in the room: ServiceNow isn't a vector database.
To make this work without relying on third-party integrations, we need to build vector indexing capabilities right into the platform's native architecture. Enter the Hierarchical Navigable Small World (HNSW) algorithm. HNSW is the gold standard for approximate nearest neighbor (ANN) searches, and building this multi-layered graph framework from scratch is how we turn ServiceNow into an intelligent search engine.

Before starting the HNSW algorithm, let's discuss about storing these chunks and vector embeddings. In python we will use sqlite with the following schema:

id INTEGER PRIMARY KEY AUTOINCREMENT,
chunk_text TEXT,
embedding_blob BLOB

To keep the construction phase lightweight, we build the HNSW graph in memory using only the node ids as pointers, rather than hauling around massive raw vector arrays. Once the ingestion pipeline completes, we serialize and dump the finalized graph topology into a local JSON file. When a search query is initiated, we simply load this JSON index back into memory—enabling lightning-fast routing without any heavy computational overhead.

Implementing this on the ServiceNow platform introduces a classic engineering constraint: execution memory and processing resources are strictly limited. Maintaining a massive, volatile graph in memory is completely unfeasible.

To adapt, we must pivot our architecture from an in-memory structure to a persistence-first design. While text chunks and vector embeddings are stored cleanly in a custom u_embeddings table, we will offload the entire HNSW graph topology into a relational schema using a second custom table, u_hnsw_relations. This table acts as our persistent graph ledger, structured with the following schema:

Number: ServiceNow auto-increment id
u_base_node: current node id
u_neighbour_node: Neighbour node id
u_layer: Layer in which current node belongs

Let's say node A has the following neighbour nodes: B, C, D, E.
We will store the details in u_hnsw_relations like this:

Number	u_base_node	u_neighbour_node
ID 1	A	B
ID 2	A	C
ID 3	A	D
ID 4	A	E

This way our graph will be stored in a table and we can use GlideRecord to traverse it.

So let's dive deep into HNSW algorithm, shall we?

Hierarchical Navigable Small World (HNSW):
Hierarchical Navigable Small World (HNSW) is a state-of-the-art algorithm used for fast Approximate Nearest Neighbor (ANN) searches. It structures high-dimensional data into a multi-layer graph, allowing search queries to skip large sections of the dataset and quickly zero in on the most similar matches.

Multi-Layered Architecture: It organizes data vectors into a multi-layered graph, mimicking a skip-list where the top layers have fewer, far-apart nodes and the bottom layer contains every single vector.
The Entry Point: Search begins at the top layer from a designated entry node, calculating the distance to its neighbors to find the closest point to the user's query vector.
Coarse Routing: It traverses the sparse top layers using large, fast leaps to quickly narrow down the general region of the target data space.
Dropping Down Layers: Once a local minimum is hit (no closer neighbors are found on that layer), the search drops down to the corresponding node in the next layer down.
Fine-Grained Navigation: This zoom-in process repeats across denser layers, refining the search radius with smaller, more precise steps.
The Ground Layer (Layer 0): On the bottom layer, it conducts a final local search among highly clustered data points to pinpoint the exact nearest neighbors.
Insertion & Probabilistic Decay: New vectors are assigned a maximum layer using a probability decay formula, ensuring most sit at the bottom while only a few act as high-level entry markers.
Heuristic Linkage: When nodes are placed, they connect to their 'M' closest neighbors, balancing fast global routing with tight local clustering to maintain the "small world" effect.

For this HNSW implementation, we will need some helper functions, so lets talk about them first.

Dot Product:
The dot product (also known as the scalar product) is an algebraic operation that takes two equal-length sequences of numbers and returns a single scalar value.
If you have two $n$-dimensional vectors, a and b, the dot product is the sum of the products of their corresponding components:

a \cdot b = i = 1 \sum n a_{i} b_{i} = a_{1} b_{1} + a_{2} b_{2} + \dots + a_{n} b_{n}

Python:

total_sum = 0
for i in range(len(vec_a)):
    total_sum += vec_a[i] * vec_b[i]
return total_sum

ServiceNow:

let totalSum = 0;
let vecLen = Math.min(vecA.length, vecB.length);
for (i = 0; i < vecLen; i++) {
     totalSum += vecA[i] * vecB[i];
}
return totalSum;

Magnitude of vector:
The magnitude of a vector is its length or size. It represents the total distance from the vector's tail (starting point) to its head (terminal point), regardless of the direction it is pointing.

Because it represents a length, the magnitude is always a scalar (a single number) and is always non-negative.
For a vector with any number of components

a = [a_{1}, a_{2}, \dots, a_{n}]

the madnitude will be:

∣ a ∣ = a_{12} + a_{22} + \dots + a_{n 2}

Python:

return math.sqrt(sum(x**2 for x in vec))

ServiceNow:

return Math.sqrt(vec.reduce((sum, x) => sum + x ** 2, 0));

Cosine Similarity:
Cosine similarity is a metric used to measure how similar two vectors are, irrespective of their size. It measures the cosine of the angle between two vectors projected in a multi-dimensional space.

Instead of measuring the distance between two points (like Euclidean distance), cosine similarity measures the direction the vectors are pointing.

Cosine Similarity = cos (θ) = \frac{a \cdot b}{∣ a ∣∣ b ∣}

Python:

mag_a = magnitude(vec_a)
mag_b = magnitude(vec_b)

if mag_a == 0 or mag_b == 0:
    return 0.0

return dot_product(vec_a, vec_b) / (mag_a * mag_b)

ServiceNow:

let magA = this.magnitude(vecA);
let magB = this.magnitude(vecB);

if (magA == 0 || magB == 0) return 0.0;
return this.dotProduct(vecA, vecB) / (magA * magB);

Cosine Distance:
Cosine distance is a metric that measures the angular disagreement or dissimilarity between two vectors. It is the direct complement to cosine similarity.

While cosine similarity tells you how close two vectors are pointing, cosine distance tells you how far apart they are.

Cosine Distance = 1 - Cosine Similarity

Python:

return 1.0 - cosine_similarity(vec_a, vec_b)

ServiceNow:

return 1.0 - this.cosineSimilarity(vec_a, vec_b);

Now that we have all our helpers, let's start designing our main HNSW Class.

Our HNSW class will have following properties:

dim: Indicates dimension of the vector i.e. 384 in this case.
conn: Database connection object only for python implementation.
M: Maximum number of connections for a node to neighbour nodes in a particular layer.
M0: Maximum number of connections for a node to neighbour nodes in 0th layer.
ef_construction: Size of the dynamic candidate list used to evaluate neighbours while the index is being built.
ef_search: Size of the dynamic candidate list used to evaluate neighbours while the graph is being searched.
mL: Normalization factor used to determine the maximum layer assignment for a newly inserted vector. $m L = \frac{1.0}{ln ( M )}$
enter_node: Starting point of the graph.
max_layer: Maximum number of layers our graph currently has.
nodes: Map to hold node indexes only in Python implementation.

Python:

class ManualHNSW:
    def __init__(self, conn, dim=384, M=16, ef_construction=200, ef_search=128):
        self.dim = dim
        self.conn = conn
        self.M = M
        self.M0 = 2 * M
        self.ef_construction = ef_construction
        self.ef_search = ef_search

        self.mL = 1.0 / math.log(M)

        self.nodes = {}
        self.total_nodes = 0

        self.enter_node = None
        self.max_layer = -1

ServiceNow:
We will maintain the graph properties in different system properties for easy access.

var RAGUtil = Class.create();
RAGUtil.prototype = Object.extendsObject(AbstractAjaxProcessor, {

    initialize: function() {
        this.M = parseInt(gs.getProperty("rag.max_conn"), 10);
        this.M0 = 2 * this.M;
        this.ef_construct = parseInt(gs.getProperty("rag.ef_construct"), 10);
        this.ef_search = parseInt(gs.getProperty("rag.ef_search"), 10);
        this.enter_node = gs.getProperty("rag.enter_node") || "";
        this.max_layer = parseInt(gs.getProperty("rag.max_layer") || "0", 10);
        this.mL = 1 / Math.log(this.M);
        this.nodes = {};

    },
    type: 'RAGUtil'
});

How Nodes Choose Their Layers:

Every node we ingest lives on the bottom layer (Layer 0), but we use a probability formula to decide if a node gets promoted to higher levels. This creates an architecture where the very top layer contains only a handful of nodes with incredibly wide, sparse connections.

As you move downward, the network of connections becomes increasingly dense. We build it this way on purpose: it mimics an express highway system. The top layers let you take huge, sweeping leaps across the data space, while the lower layers act as local streets that guide you right to the exact data point.

We will use the following decay probability formula to determine the layer for the new Node/vector:

layer = ⌊ - ln (random ()) \times m_{L} ⌋

where

m_{L} = \frac{1}{ln ( M )}

If your maximum connections M = 16:

m_{L} = 1.0/ ma t h . l o g (16) \approx 0.3606

Now, look at what layer a node gets assigned depending on what random number (r) it draws:

Random Draw (r)	Math: -math.log(r)	Multiplied by mL (0.3606)	Final Integer Layer
0.75 (Common draw)	0.287	0.103	Layer 0
0.30 (Average draw)	1.203	0.434	Layer 0
0.05 (Rare draw)	2.995	1.080	Layer 1
0.001 (Extremely rare)	6.907	2.490	Layer 2

Python:

def _get_random_layer(self):
    r = random.random()
    if r == 0:
        r = 0.0000001

    return int(-math.log(r) * self.mL)

ServiceNow:

getRandomLayer: function() {
    let r = Math.random();
    if (r == 0) r = 0.0000001;
    return Math.floor(-Math.log(r) * this.mL);
}

Now it's time to address the elephant in the room, the search mechanism. This is the core of the HNSW algorithm which is essential for both ingestion and retrieval phase. We will use two type of search:

Single candidate greedy search: We will find out the node closest to our query in a layer and drill down to the next layer with its connection. This will help us to reach the ground layer as soon as possible.
Multi-candidate beam search: In this case we will maintain a candidate pool and perform granular search among the neighbours.

Single-candidate greedy search:
Once we are on a specific layer of our HNSW graph, how do we actually find the node closest to our user's query? We use a Greedy Search strategy.

Think of this function like a hiker trying to find the lowest point in a valley during a foggy day—they simply look at their immediate surroundings, take a step in whatever direction goes downward, and repeat until every direction leads back uphill.

Here is exactly how the logic executes:

The Baseline: It calculates the cosine_distance from the query_vector to our starting point (enter_node) and sets that as the distance to beat.
Checking the Neighborhood: It fetches all the connected neighbours for the current node restricted strictly to the active graph layer.
The "Greedy" Shift: It iterates through those neighbors. The moment it finds a neighbor that is mathematically closer to the query vector than our current node, it updates curr_node to that neighbor.
The Loop Break: It keeps hopping from neighbor to closer neighbor in a while True loop. The moment a full scan of the neighborhood yields no closer points (if not changed), the algorithm knows it has hit a local minimum for this layer and returns the closest node found.

This local winner then becomes the entry point for the next layer down, systematically zooming in on the perfect data match!

Python:

def _search_layer(self, query_vector, enter_node, layer):
    curr_node = enter_node
    curr_dist = self.cosine_distance(query_vector, curr_node)
    while True:
        changed = False
        neighbours = self.nodes.get(curr_node, {}).get(layer, [])
        for neighbour in neighbours:
            neighbour_dist = self.cosine_distance(query_vector, neighbour)
            if neighbour_dist < curr_dist:
                curr_dist = neighbour_dist
                curr_node = neighbour
                changed = True

        if not changed:
            break

    return curr_node

ServiceNow:

searchLayer: function(query_vector, enter_node, layer) {
    var curr_node = enter_node;
    var curr_dist = this.cosineDistance(query_vector, curr_node);

    while (true) {
        var changed = false;

        var neighbourGr = new GlideRecord("u_hnsw_relations");

        neighbourGr.addQuery("u_base_node", curr_node);
        neighbourGr.addQuery("u_layer", layer);
        neighbourGr.query();

        while (neighbourGr.next()) {
            let neighbour_dist = this.cosineDistance(query_vector, neighbourGr.getValue("u_neighbour_node"));

            if (neighbour_dist < curr_dist) {
                curr_dist = neighbour_dist;
                curr_node = neighbourGr.getValue("u_neighbour_node");
                changed = true;
            }
        }

        if (!changed) {
            break;
        }
    }

    return curr_node;
}

Multi-candidate beam search:
Our previous search function was completely greedy—it hopped to one best neighbour and threw away the rest. That runs a high risk of getting stuck in a local dead end.

This upgraded method introduces ef (Exploration Factor). Instead of tracking just one node, we are now maintaining a dynamic "pool" of the top-ef closest nodes. Think of it like changing our search strategy from a single scout into a well-coordinated search party. We will use this search in layer 0 to get most accurate result.

Here is the underlying mechanics of how this tracking works:

The Two Pools: We initialize a visited set to avoid back-tracking, a candidates queue (nodes we still need to explore), and v_pool (our tracking pool capped at size ef that keeps our absolute best results).
The Priority Sort: In the while loop, we sort our candidates by distance so we always pop and inspect the closest unexplored node (curr_node) first.
The Smart Early-Exit: If the node we are about to inspect is already further away than the worst node in our v_pool (if curr_dist > v_pool[-1][1]), the algorithm stops. We've hit a point of diminishing returns.
Evaluating Neighbours: We loop through the neighbors of our current node. If a neighbour is closer than the current worst node in our pool, OR if our pool hasn't reached the ef limit yet, we add that neighbor to both our candidates and our v_pool.
Maintaining the Cap: We immediately re-sort v_pool and use v_pool.pop() to kick out the furthest element if the size exceeds ef. This keeps the pool locked at our optimal search size.

By the time the loop ends, the function returns a list of the top-ef closest nodes on this layer, massively increasing our search accuracy and preventing the algorithm from getting blindsided by a bad graph split!

Python:

def _search_layer_ef(self, query_vector, enter_node, layer, ef):
    visited = {enter_node}
    init_dist = self.cosine_distance(query_vector, enter_node)

    v_pool = [(enter_node, init_dist)]
    candidates = [(enter_node, init_dist)]

    while candidates:
        candidates.sort(key=lambda x: x[1])
        curr_node, curr_dist = candidates.pop(0)

        if curr_dist > v_pool[-1][1]:
            break

        neighbors = self.nodes.get(curr_node, {}).get(layer, [])
        for neighbor in neighbors:
            if neighbor not in visited:
                visited.add(neighbor)

                neighbor_dist = self.cosine_distance(query_vector, neighbor)

                if neighbor_dist < v_pool[-1][1] or len(v_pool) < ef:
                    candidates.append((neighbor, neighbor_dist))
                    v_pool.append((neighbor, neighbor_dist))
                    v_pool.sort(key=lambda x: x[1])

                    if len(v_pool) > ef:
                        v_pool.pop()

    return [node_id for node_id, _ in v_pool]

ServiceNow:

searchLayerEf: function(query_vector, enter_node, layer, ef) {
    var enter_node_id = enter_node.toString();

    var visited = {};
    visited[enter_node_id] = true;

    var init_dist = this.cosineDistance(query_vector, enter_node_id);

    var v_pool = [{
        id: enter_node_id,
        dist: init_dist
    }];
    var candidates = [{
        id: enter_node_id,
        dist: init_dist
    }];

    while (candidates.length > 0) {
        candidates.sort(function(a, b) {
            return a.dist - b.dist;
        });

        var curr = candidates.shift();
        var curr_node_id = curr.id;
        var curr_dist = curr.dist;

        if (curr_dist > v_pool[v_pool.length - 1].dist) {
            break;
        }

        var neighbourGr = new GlideRecord("u_hnsw_relations");
        neighbourGr.addQuery("u_base_node", curr_node_id);
        neighbourGr.addQuery("u_layer", layer);
        neighbourGr.query();

        while (neighbourGr.next()) {
            var neighbour_id = neighbourGr.getValue("u_neighbour_node");

            if (!neighbour_id) {
                gs.warn("Skipping null neighbour for base_node=" + curr_node_id + " layer=" + layer);
                continue;
            }

            if (!visited[neighbour_id]) {
                visited[neighbour_id] = true;

                var neighbour_dist = this.cosineDistance(query_vector, neighbour_id);
                var far_dist = v_pool[v_pool.length - 1].dist;

                if (neighbour_dist < far_dist || v_pool.length < ef) {
                    var newCandidate = {
                        id: neighbour_id,
                        dist: neighbour_dist
                    };

                    candidates.push(newCandidate);
                    v_pool.push(newCandidate);

                    v_pool.sort(function(a, b) {
                        return a.dist - b.dist;
                    });

                    if (v_pool.length > ef) {
                        v_pool.pop();
                    }
                }
            }
        }
    }

    var resultNodes = [];
    for (var i = 0; i < v_pool.length; i++) {
        resultNodes.push(v_pool[i].id);
    }

    return resultNodes;
}

Now it's time to design our insert function for the graph. The insert function is the core engine that constructs our multi-layered graph. When a new vector arrives, it undergoes a structured three-phase pipeline:

Phase 1: First-Node Initialization
If the graph is completely empty, the algorithm sets up a clean slate. It designates this initial node as the global entry point (self.enter_node), assigns it to the base layer (Layer 0), and gracefully exits.
Phase 2: Layer Allocation & Fast-Pass Routing
If the graph already contains data, the node is probabilistically assigned a maximum height (insert_layer). Before making any connections, the algorithm executes a fast-pass global search starting from the very top of the existing graph down to the node's designated insert_layer. This rapidly brings us to the correct spatial neighborhood without altering any graph links.
Phase 3: Multi-Layer Linkage & Pruning
Once the entry point drops down into the node's target layers, the actual insertion work begins. Moving from the insert_layer down to Layer 0, the function:

Utilizes our advanced _search_layer_ef search party to locate the closest candidate neighbors on that layer.

Binds the new node to these neighbors bidirectionally.

Evaluates if the new links exceed our allowed thresholds (M or M0). If a node becomes overly clustered, it triggers a pruning heuristic to slice away the furthest connections, keeping our routing layout tightly optimized.

If our new node happens to roll a random layer higher than anything seen before, the algorithm updates the graph’s global max_layer and crowns this new node as our new global entry point!

Python:

def insert(self, new_node_id, new_vector):
    self.total_nodes += 1
    # if graph is empty
    if not self.nodes:
        self.enter_node = new_node_id
        self.nodes[new_node_id] = {0: []}
        self.max_layer = 0
        return

    # if graph is not empty
    insert_layer = self._get_random_layer()
    self.nodes[new_node_id] = {}
    for i in range(0, insert_layer + 1):
        self.nodes[new_node_id][i] = []

    curr_obj = self.enter_node

    for l in range(self.max_layer, insert_layer, -1):
        curr_obj = self._search_layer(new_vector, curr_obj, l)

    for l in range(min(self.max_layer, insert_layer), -1, -1):
        candidates = self._search_layer_ef(
            new_vector, curr_obj, l, self.ef_construction
        )
        curr_max_links = self.M0 if l == 0 else self.M

        top_neighbors = self.get_top_k(new_vector, candidates, curr_max_links)
        curr_obj = top_neighbors[0]

        for closest_node in top_neighbors:
            self.nodes[new_node_id][l].append(closest_node)
            self.nodes[closest_node][l].append(new_node_id)

            if len(self.nodes[closest_node][l]) > curr_max_links:
                self.nodes[closest_node][l] = self.prune_to_max_connection(
                    closest_node, l, curr_max_links
                )

        if len(self.nodes[new_node_id][l]) > curr_max_links:
            self.nodes[new_node_id][l] = self.prune_to_max_connection(
                new_node_id, l, curr_max_links
            )

    if insert_layer > self.max_layer:
        self.max_layer = insert_layer
        self.enter_node = new_node_id

ServiceNow:

insert: function(new_node_id, new_vector) {
    new_node_id = new_node_id.toString();

    if (!this.enter_node) {
        this.enter_node = new_node_id;
        this.max_layer = 0;
        gs.setProperty("rag.enter_node", this.enter_node);
        gs.setProperty("rag.max_layer", "0");
        this._createNodePlaceholder(new_node_id);
        return;
    }

    var insert_layer = this.getRandomLayer();

    var curr_obj = this.enter_node.toString();

    for (var l = this.max_layer; l > insert_layer; l--) {
        curr_obj = this.searchLayer(new_vector, curr_obj, l);
    }

    var start_layer = Math.min(this.max_layer, insert_layer);
    for (var l = start_layer; l >= 0; l--) {

        var candidates = this.searchLayerEf(new_vector, curr_obj, l, this.ef_construct);
        var curr_max_links = (l === 0) ? this.M0 : this.M;

        var top_neighbors = this.getTopK(new_vector, candidates, curr_max_links);
        if (top_neighbors.length > 0) {
            curr_obj = top_neighbors[0];
        }

        for (var i = 0; i < top_neighbors.length; i++) {
            var closest_node = top_neighbors[i].toString();

            this._addRelation(new_node_id, closest_node, l);
            this._addRelation(closest_node, new_node_id, l);

            if (this._getLinkCount(closest_node, l) > curr_max_links) {
                this.pruneToMaxConnection(closest_node, l, curr_max_links);
            }
        }

        if (this._getLinkCount(new_node_id, l) > curr_max_links) {
            this.pruneToMaxConnection(new_node_id, l, curr_max_links);
        }
    }

    if (insert_layer > this.max_layer) {
        this.max_layer = insert_layer;
        this.enter_node = new_node_id;
        gs.setProperty("rag.enter_node", this.enter_node);
        gs.setProperty("rag.max_layer", this.max_layer.toString());
    }
}

Our ingestion pipeline is 90% done, we just need to put together all the things. Remember how our pipeline will work:

Read text
Chunk it
create embeddings
store in db

Python:

text = readFile(file_path)
chunks = createChunks(text, chunk_size=250, overlap=50)
embeddings = createEmbeddings(chunks)
cursor = conn.cursor()
for i in range(len(chunk_list)):
    format_string = "<384f"
    embedding_blob = struct.pack(format_string, *raw_embeddings[i])
    cursor.execute(
        "INSERT INTO documents (chunk_text, embedding_blob) VALUES (?, ?)",
        (chunk_list[i], embedding_blob),
    )

    row_id = cursor.lastrowid
    hnsw.insert(row_id, raw_embeddings[i])
conn.commit()

ServiceNow:
Note: We will be using Scheduled job for ingestion in ServiceNow as ingestion job takes much time. We need to run the job in background as the time limit may exceed if we try to run it in foreground and we will end up with broken relations between nodes or may end up in partial ingestion.

gs.setProperty("rag.enter_node", "");
gs.setProperty("rag.max_layer", 0);

var embGr = new GlideRecord("u_embeddings");
var relGr = new GlideRecord("u_hnsw_relations");
embGr.deleteMultiple();
relGr.deleteMultiple();


var kbGr = new GlideRecord("kb_knowledge");
kbGr.addEncodedQuery("workflow_state=published");
kbGr.query();

while (kbGr.next()) {
    try {
        var articleNumber = kbGr.number.toString();
        gs.info("HNSW Ingestion: Attempting processing for " + articleNumber);

        var text = kbGr.getDisplayValue('text');

        if (!text || text.trim() === "") {
            gs.warn("HNSW Ingestion: Skipping article " + articleNumber + " because the text field is empty.");
            continue; 
        }

        var rag = new RAGUtil(); 

        var chunkList = rag.createChunks(text, 200, 20);

        if (!chunkList || chunkList.length === 0) {
            gs.warn("HNSW Ingestion: Chunk generation returned 0 blocks for article " + articleNumber);
            continue;
        }

        var rawEmbeddings = rag.createEmbeddings(chunkList);

        if (!rawEmbeddings || rawEmbeddings.length === 0) {
            gs.warn("HNSW Ingestion: Embedding generation returned 0 vectors for article " + articleNumber);
            continue;
        }

        rag.storeEmbeddings(chunkList, rawEmbeddings);

        gs.info("HNSW Ingestion: Successfully processed and committed " + articleNumber);

    } catch (articleError) {
        gs.error("HNSW Ingestion FATAL EXCEPTION on article " + kbGr.number.toString() + ". Error Details: " + articleError.toString());
    }
}

gs.info("HNSW Ingestion: Completed successfully!")

Yay! our ingestion pipeline is ready.

We have all the important components at our disposal so implementing Retrieval mechanism will be a piece of cake! Without further ado let's start designing the retrieval segment.

Data Retrieval:
Main part of retrieval is the search. We have our Single-candidate greedy and multi-candidate beam search already implemented, we just need to put them together.
The strategy will be almost same as the insert method, we will use the greedy search to quickly reach the ground layer, then conduct a beam search to get top candidate.

Python:

def search(self, query, k):
    query_vector = createQueryEmbedding(query)

    if not self.enter_node or not self.nodes:
        return []

    curr_obj = self.enter_node

    for l in range(self.max_layer, 0, -1):
        curr_obj = self._search_layer(query_vector, curr_obj, l)

    candidates = self._search_layer_ef(query_vector, curr_obj, 0, self.ef_search)

    top_k_ids = self.get_top_k(query_vector, candidates, k)

    return [self.get_chunk(node_id) for node_id in top_k_ids]

ServiceNow:

search: function(query, k) {
    var query_vector = this.createQueryEmbedding(query);

    if (!this.enter_node) {
        return [];
    }

    var curr_obj = this.enter_node.toString();
    var current_max_layer = parseInt(this.max_layer, 10) || 0;

    for (var l = current_max_layer; l > 0; l--) {
        curr_obj = this.searchLayer(query_vector, curr_obj, l);
    }

    var efSearchLimit = parseInt(this.ef_search, 10) || 16;
    var candidates = this.searchLayerEf(query_vector, curr_obj, 0, efSearchLimit);

    var top_k_ids = this.getTopK(query_vector, candidates, k);

    var finalChunks = [];
    for (var i = 0; i < top_k_ids.length; i++) {
        var chunkText = this.getChunk(top_k_ids[i]);
        if (chunkText) {
            finalChunks.push(chunkText);
        }
    }

    return finalChunks;
}

And... that's it! We have our retrieval mechanism in place!
Phew! it's a loooong blog, isn't it? Just bear with me a bit more, we are almost done, I promise...

Generation:
We have our ingestion and retrieval mechanism in place. We we can use them to retrieve data from our data source and feed them to the LLM. Our Ingestion mechanism will run once and store all the chunks and embeddings in the databases and graphs. Our retrieval mechanism will run every time a query arrives. In ServiceNow we can use it in any server-side script.

Python:

conn = getSqliteConnection(clear_on_start=False)
    hnsw_graph = ManualHNSW(conn)
    hnsw_graph.load_from_file("hnsw.json")

    print("[+] Generating query embeddings...")

    # creating embeddings for the user query
    query_embeddings = createQueryEmbedding(query=query)
    print("[✓] Query embeddings generated...")

    print("[+] Started similarity search...")

    # Querying HNSW(similarity search)
    retrieved_chunks = hnsw_graph.search(query, 5)

    conn.close()

    print("[✓] Similar chunks retrieved...")

    # Creating context string to pass to llm
    context = "\n---\n".join(retrieved_chunks)

    # print("Context:\n\n", context)

    system_prompt = (
            "You are a helpful assistant. Answer the user's question using ONLY the provided text context. "
            "If the answer cannot be found in the context, say 'I cannot find the answer in the document.' "
            "Do not make up information or use outside knowledge."
        )

    user_prompt = f"""Context:
{context}

Question: {query}
Answer:"""

    print("[+] Prompting LLM...")

    # Calling GROQ Api
    response = callLLM(system_prompt=system_prompt, user_prompt=user_prompt)
    print("[✓] Response received...")

    print("\n--- Response ---\n")
    print(response)

ServiceNow:

var rag = new RAGUtil();
var query = "How to set automatic replies in Microsoft Outlook?";

var topChunks = rag.search(query, 3);

var context = topChunks.join("\n---\n");

var systemPrompt = `You are a helpful assistant. Answer the user's question using ONLY the provided text context.
If the answer cannot be found in the context, say 'I cannot find the answer in the document.'
Do not make up information or use outside knowledge.`

var userPrompt = `Context:
${context}
Question: ${query}
Answer:`

gs.print(rag.callLLM(systemPrompt, userPrompt));

And we are done!

Building a RAG pipeline from the ground up — without leaning on LangChain, ChromaDB, or any dedicated vector database — is genuinely one of the most enlightening exercises you can put yourself through as a developer. Every abstraction you strip away reveals a layer of the machinery that high-level frameworks quietly handle for you.
What makes this implementation particularly interesting is the ServiceNow port. Adapting an in-memory, file-backed Python prototype to a resource-constrained, persistence-first enterprise platform forced real architectural decisions: swapping JSON files for relational graph storage in u_hnsw_relations, replacing volatile in-memory node maps with GlideRecord traversals, and running ingestion as a scheduled background job to sidestep execution timeouts. These aren't workarounds — they are the kind of pragmatic engineering trade-offs that separate prototype thinking from production thinking.
At its core, what we built here is a fully functional semantic search engine living inside a platform that was never designed to be one. No third-party vector databases, no bloated dependencies — just chunked text, float arrays, a hand-rolled HNSW graph, and a well-crafted prompt.
The takeaway is simple: understanding the fundamentals gives you the freedom to build anywhere. Whether you are working in Python, ServiceNow, or something entirely different, the mechanics of RAG remain the same. Master the core, and the platform becomes just an implementation detail.

DISCLAIMER: This was just an experiment from my end to check how much effort it will take to implement something like a RAG system. ServiceNow Platform is not built for compute heavy implementations like this and may degrade system performance if a very large corpus is used.

All code used for this blog available here:

mainak55512 / RAG-in-SERVICENOW

Why not Language Specific SLMs as coding agents

Mainak Bhattacharjee — Sun, 10 May 2026 06:28:28 +0000

Coding should be free, private, secure, and accessible offline — no exceptions. It has always been one of humanity's greatest equalizers, a skill that anyone with a cheap laptop and curiosity could pick up and use to build something real. Putting it behind a paywall doesn't just create inconvenience — it creates inequality. It turns a democratic craft into a privilege, and that is a direction we should actively resist.

There is a better path. Instead of chasing ever-larger general-purpose models that only run on expensive infrastructure, we should be building Language Specific Small Language Models — compact, focused models that are genuine experts in a single programming language.

The Core Logic is Sound:

A general coding SLM still carries baggage — it knows 50+ languages, frameworks, paradigms. But most developers work in 1 or 2 languages most of their career. So why should their local model waste capacity on the rest?

A Python-only SLM doesn't need to know:

Rust's borrow checker
Java's verbose OOP patterns
C++ memory management
Ruby's metaprogramming

Strip all that out and suddenly your model can be dramatically smaller while being dramatically better at the one thing it does.

What Language-Specific SLMs Could Look Like:

PyroLM (Python) — 500M params, knows Python, pip, and common libs, runs on 2GB RAM

WebLM (JS/TS) — 500M params, knows JS, TS, Node, and the npm ecosystem, runs on 2GB RAM

QueryLM (SQL) — 300M params, knows SQL dialects and query optimization, runs on 1GB RAM

ShellLM (Bash) — 200M params, knows shell scripting and Linux tools, runs on 512MB RAM

RustLM — 600M params, knows Rust, Cargo, and the ownership model, runs on 2GB RAM

These could genuinely run on very low-end hardware — even phones.

The Deeper Advantages Nobody Talks About:

Deeper ecosystem knowledge A Python-only model can deeply learn not just the language but the entire ecosystem — NumPy, Pandas, FastAPI, Django, pytest patterns, Pythonic idioms, PEP standards. A general model skims all of this. A specialist model lives in it.
Better error understanding Python errors are different from JS errors. A Python-specific model trained heavily on debugging patterns, stack traces, and common mistakes would catch errors a general model misses.
Version awareness Python 3.12 is different from 3.8. A language-specific model can be trained to be deeply aware of version differences, deprecations, and migration paths — something general models handle poorly.
Smaller = faster = better UX A 300MB model responds almost instantly even on weak hardware. That responsiveness changes how people use it — quick inline suggestions, real-time checks, not waiting 10 seconds for a response.
Easier to update and maintain When Python 3.14 comes out, you retrain or fine-tune just the Python model. You don't have to touch anything else. Much more maintainable long term.

What Training Data Would Look Like:

This is where language-specific models have a massive advantage:

Curated, verified repositories in that language only
Official documentation deeply embedded
Common error → fix pairs scraped from Stack Overflow
Code review comments teaching idiomatic patterns
Test suites teaching what "correct" looks like
Changelogs and migration guides for version awareness
Quality over quantity. 10GB of excellent Python code beats 1TB of random multilingual code for this purpose.

The Honest Challenges:

Context still matters Real projects mix things. A Python backend developer still writes YAML configs, SQL queries, Bash scripts, Dockerfiles. A pure Python model is blind to all of that. You'd either need multiple models running together or accept that limitation.
The switching friction Developers would need to switch models when switching languages. That's a workflow problem that needs good tooling to solve.
Who builds and maintains it Language communities themselves are the natural answer — the Python Software Foundation, the Rust Foundation, the JS community. But these organizations have historically focused on the language, not AI tooling. That culture would need to shift.

The Most Exciting Possibility:

Imagine this stack:

Your laptop (4GB RAM)
├── Router model (tiny) → detects what language you're writing
├── PyroLM → activates when you're in .py files
├── QueryLM → activates when you're writing SQL
├── ShellLM → activates in terminal
└── All offline. All free. All fast. Forever

A smart VS Code extension could handle the routing invisibly. You'd never think about which model is running — it just works, privately, for free, on any machine.

But Academic prestige in AI research goes to bigger, more general models. Publishing "we made a 500M parameter Python-only model" doesn't win the same attention as "we made a 70B model that beats GPT-4." The research community's incentives push toward general and large, not specific and small.

This idea deserves to be built. It's practical, technically sound, and solves a real access problem. The open source community — especially individual language communities — are the ones who could actually make this happen. Someone just needs to start.

What do you think?

From Regex Rampage to Lazy Bliss: My rjq Performance Adventure

Mainak Bhattacharjee — Sat, 12 Oct 2024 16:19:08 +0000

Hey there, fellow Rustaceans 🦀!

I've been building a JSON filter tool called rjq, inspired by the awesome jq. But things took a turn for the worse when I hit a performance wall during lexing. The culprit? Compiling regular expressions in a hot loop . It turns out, regexes are like hungry hippos – they chomp up performance if you're not careful!
Here's the story of how I tamed the regex beast and saved my program from a slow, sluggish fate:

The Regex Rampage 🦖:

At first, I naively compiled the regex patterns within the lexing loop. This meant every iteration involved creating a brand new regex object. Think of it like baking a whole new pizza for every bite – inefficient, right? This constant creation caused a major performance bottleneck i.e. ~80% execution time was consumed by this.

The Lazylock Solution 🧙‍♂️:

Thankfully, the Rust gods (and some helpful folks on the r/Rust subreddit) pointed me towards lazy_static and a technique called lazy initialization. This magic combo allowed me to compile the regex only once and store it in a thread-safe location using a LazyLock. Now, it's like having a box of pizza ready with a fresh slices whenever you need it – much more efficient!

The Lazy Bliss ✨:

The impact was phenomenal! Performance soared, and my lexing code became as smooth as butter . No more regex rampage, just happy filtering .
Want to See the Code?
Curious about the details? Head over to my GitHub repo for rjq: https://github.com/mainak55512/rjq

Lessons Learned 📚:

Regex compilation can be expensive, avoid hot loops!
Embrace lazy initialization for performance gains.
There's always a better way to do things in Rust (and life!)

So, the next time you encounter a performance bottleneck, remember – there might be a lazy solution waiting to be discovered!

P.S. If you have any other tips or tricks for optimizing JSON filtering in Rust, leave a comment below!

But wait, there's more!

Let's dive deeper into the technical aspects of this adventure.
Understanding lazy_static and LazyLock

lazy_static: This macro provides a way to declare static variables that are initialized only once, even in a multi-threaded environment.
LazyLock: This is a type provided by the lazy_static crate that ensures thread-safety during initialization.

Here's a simplified example of how I used these to optimize the regex compilation in rjq:

Outside the hot loop:

static MATCH_NUMBER: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"^\d+\.?\d+").unwrap());

...and so on

Inside the hot loop:

    if MATCH_NUMBER.is_match(&source_string[cursor..]) {
        match MATCH_NUMBER
            .find(&source_string[cursor..])
            .map(|x| x.as_str())
        {
            Some(val) => {
                cursor += val.len();
                token_array.push_back(token(TokenType::NUMBER, val.to_string()));
            }
            None => (),
        }
    } else if ... so on

As you can see, the MATCH_NUMBER variable is declared using LazyLock, and it's initialized only once when the code is first executed. The LazyLock within the code ensures that the initialization is thread-safe.

Additional Performance Tips

Profiling: Use tools like perf or cargo-flamegraph to identify other performance bottlenecks in your code.
Data Structures: Choose appropriate data structures for your use case. For example, consider using HashMap for efficient lookups.
Algorithms: Optimize algorithms to reduce computational complexity.
Memory Management: Be mindful of memory allocations and deallocations.

By following these tips and leveraging techniques like lazy initialization, you can significantly improve the performance of your Rust applications.

Happy coding 🎉!

Introducing rjq: A Fast and Lightweight CLI JSON Filtering Tool

Mainak Bhattacharjee — Fri, 11 Oct 2024 06:05:31 +0000

In the world of data manipulation, JSON has become a ubiquitous format, but filtering and querying JSON data can be cumbersome without the right tools. Enter rjq, a command-line JSON filtering tool developed in Rust🦀, designed to be a performant and lightweight alternative to the popular jq tool.

The Motivation Behind rjq

rjq began as a hobby project, driven by a desire to create a tool that prioritizes performance and simplicity. With a keen focus on providing a robust alternative to jq, rjq is compatible with both Linux and Windows, making it accessible to a wider audience.
Key Features

Performance:

rjq has been optimized for speed, running nearly 2x faster than jq when tested on a Linux machine with 4GB RAM and an Intel i3 6th Gen processor. This performance boost can significantly enhance workflows, especially for users dealing with large datasets.

Simplicity:

The query structure of rjq is designed to be intuitive. Writing queries feels akin to crafting simple conditional statements in any programming language, which lowers the barrier to entry for new users.

Lightweight:

With a minimalistic approach, rjq ensures that users can quickly load and filter JSON data without unnecessary overhead.

How to Use rjq

Using rjq is straightforward. You can load JSON data from a file using the --load flag, or you can pipe input directly into the tool. Here are some usage examples:


rjq --load="test.json" --query="<query string>" --params="<comma separated parameter list>"

Alternatively, you can pipe JSON output from other commands:


stto --json cpython | rjq --query="<query string>" --params="<comma separated parameter list>"

The Development Journey

The development of rjq has been a valuable learning experience, particularly in mastering the intricacies of Rust. The support from the Reddit community was instrumental in overcoming challenges. You can check out some of the discussions and insights from fellow developers in this Reddit post.

Future Plans

Looking ahead, the goal for rjq is to become the go-to choice for JSON filtering on both Linux and Windows. rjq is still in early development stages and there are plans to incorporate new features and enhance the tool's capabilities to meet the needs of users day-to-day usage. Any contribution, stars ⭐ and fork 🔗 to the rjq repo is greatly appreciated 👍.

Who Can Benefit from rjq?

rjq is tailored for:

Developers:

Those working with JSON data who need a reliable filtering tool.

Data Analysts:

Professionals seeking efficient data extraction methods.

DevOps Teams:

Teams automating data processing tasks in their workflows.

Installation

Getting started with rjq is easy. Binaries for both Linux and Windows are available in the releases section of the GitHub repository, allowing users to install the tool without hassle.

Conclusion

Whether you’re a developer, data analyst, or part of a DevOps team, rjq offers a fast, lightweight solution for filtering JSON data. With its performance, simplicity, and growing feature set, rjq is poised to become an essential tool in your data processing arsenal. Check out the GitHub repository to learn more and get started today!