Building an Agentic RAG System from Scratch
AI Agents · LLMs · Generative AI · AI
By the end of this article, you’ll have implemented an agentic RAG system twice: once from scratch to understand every moving part, and once with LlamaIndex and LangGraph to see how production frameworks encapsulate the same logic.
Everything here runs on Gemini’s free API tier through Google AI Studio, so you can follow along without a credit card. You can also use Groq’s API instead — I’ll show the swap where it matters.
What we’re building
We’re going to build a small Q&A agent over a fictional API product’s documentation — call it “ByteFlow API.” (A chatbot suggested the name when I asked it to draft some sample product documentation.) Five short docs: pricing, rate limits, error codes, authentication, and webhooks. Small enough to read in under a minute, but structured so that some questions genuinely need more than one doc at once — which is exactly what demonstrates an agentic workflow.
That last part matters most. If every test question is answerable from a single doc, traditional RAG looks just as good as agentic RAG. So our test question is deliberately multi-hop:
“I’m on the free plan and I got blocked after a bunch of requests today. What’s happening and what are my options?”
Answering that well means combining the rate-limits doc (what a block actually is), the pricing doc (what “free plan” means for limits), and the errors doc (what to actually do about it). One vector lookup against one doc won’t cover all three.
Setup
Grab a free API key from Google AI Studio, then:
pip install google-genai torch
import os
os.environ["GEMINI_API_KEY"] = "your-key-here" # or use a .env file
We’ll use gemini-embedding-001 for embeddings and gemini-3.6-flash for generation. Google renames and reshuffles Gemini model versions fairly often, so check the current list in AI Studio and swap in whatever's current — nothing else in this code depends on the specific model name.
Using Groq instead?
pip install groq
import os
os.environ["GROQ_API_KEY"] = "your-key-here" # or use a .env file
I like llama-3.3-70b-versatile for generation on Groq. Model and embedding selection for different providers is worth its own article — this one sticks to Gemini as the default, with the Groq swap shown where the generation code lives below.
Part 1 — Building the loop by hand
No framework yet. The point of this section is to make the agent loop boring and mechanical — because that’s exactly what it is once you strip the framework abstraction away.
The document set and a simple vector index
We don’t need a real vector database for five documents — a plain Python list and cosine similarity is enough. This is a demo of agentic RAG, not a vector database benchmark.
One design detail matters more than it looks like it should: the answer to our test question can’t live entirely inside one document. The three facts a complete answer needs — why you got blocked, what the actual tier numbers are, and what your options are — are deliberately split across rate_limits, pricing, and errors, with no single doc containing enough to answer alone. If they overlapped, a single embedding lookup could accidentally grab the one doc that covers everything, and the whole point of the comparison would disappear.
import os
import json
import torch
import torch.nn.functional as F
from google import genai
from google.genai import types
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
GEN_MODEL = "gemini-3.6-flash"
EMBED_MODEL = "gemini-embedding-001"
DOCS = [
{
"id": "pricing",
"text": (
"ByteFlow API pricing tiers: Free tier costs $0 per month and includes "
"100 requests per day, with no webhook access. Pro tier costs $29 per "
"month and includes 10,000 requests per day, plus webhook access. "
"Enterprise tier has custom volume pricing and is billed annually."
),
},
{
"id": "rate_limits",
"text": (
"Rate limits are enforced per API key, not per account, and reset "
"daily at midnight UTC. Exceeding your allotted requests for the day "
"returns an HTTP 429 response for the remainder of the current window. "
"Your specific daily request cap depends on your subscription tier - "
"see the pricing page for exact tier-by-tier numbers."
),
},
{
"id": "errors",
"text": (
"Error code reference: 401 means an invalid or missing API key. 429 "
"means you have exceeded your rate limit for the current window. 500 "
"means an internal server error, and requests should be retried with "
"backoff. If you're seeing repeated 429 responses, you can wait for "
"your daily window to reset, or move to a higher-volume subscription tier."
),
},
{
"id": "auth",
"text": (
"To authenticate requests, pass your API key in the Authorization header as "
"a Bearer token, e.g. 'Authorization: Bearer YOUR_API_KEY'. Keys are generated "
"from the dashboard under Settings > API Keys."
),
},
{
"id": "webhooks",
"text": (
"Webhooks let you receive real-time events when data changes, instead of "
"polling the API. Webhook access is only available on Pro tier and above. "
"Configure your endpoint URL from the dashboard under Settings > Webhooks."
),
},
]
def embed_text(text, task_type="RETRIEVAL_DOCUMENT"):
result = client.models.embed_content(
model=EMBED_MODEL,
contents=text,
config=types.EmbedContentConfig(task_type=task_type),
)
return torch.tensor(result.embeddings[0].values)
def build_index(docs):
for doc in docs:
doc["embedding"] = embed_text(doc["text"], task_type="RETRIEVAL_DOCUMENT")
return docs
def cosine_similarity(a, b):
return F.cosine_similarity(a.unsqueeze(0), b.unsqueeze(0)).item()
def retrieve(query, docs, top_k=2):
query_emb = embed_text(query, task_type="RETRIEVAL_QUERY")
scored = [(doc, cosine_similarity(query_emb, doc["embedding"])) for doc in docs]
scored.sort(key=lambda pair: pair[1], reverse=True)
return [doc for doc, score in scored[:top_k]]
def generate(prompt):
response = client.models.generate_content(model=GEN_MODEL, contents=prompt)
return response.text
Using Groq for generation instead? Swap in this version of generate (embeddings still go through Gemini, since Groq doesn't serve an embedding model):
from groq import Groq
groq_client = Groq(api_key=os.environ["GROQ_API_KEY"])
GEN_MODEL = "llama-3.3-70b-versatile"
def generate(prompt):
response = groq_client.chat.completions.create(
model=GEN_MODEL,
messages=[{"role": "user", "content": prompt}],
temperature=0.2,
)
return response.choices[0].message.content
Nothing agentic yet — this is just the standard plumbing every RAG system needs.
The traditional baseline
Here’s the traditional RAG baseline we’re improving on:
def traditional_rag(query, docs):
chunks = retrieve(query, docs, top_k=2)
context = "\n\n".join(chunk["text"] for chunk in chunks)
prompt = (
f"Answer the question using only the context below.\n\n"
f"Context:\n{context}\n\nQuestion: {query}"
)
return generate(prompt)
Embed, retrieve, generate. One shot, no questions asked about whether what came back was actually good enough.
Adding the agent loop
This is the actual point of this article. Four pieces, in order.
1. Decompose. Embedding models produce a single vector representing the entire sentence. If one sentence contains multiple intents, that vector averages them together. Splitting the query lets each embedding focus on one semantic objective, improving retrieval precision.
def decompose(query):
prompt = (
"Break the following question into 1-3 self-contained sub-questions "
"needed to answer it fully. Respond with a JSON array of strings, "
"nothing else.\n\n"
f"Question: {query}"
)
raw = generate(prompt)
cleaned = raw.strip().strip("`").replace("json", "", 1).strip()
try:
return json.loads(cleaned)
except json.JSONDecodeError:
return [query]
The try/except here matters more than it looks like it should. Even when explicitly asked for JSON only, LLMs will occasionally wrap the response in a code fence or add a stray sentence — without this fallback, that kind of formatting hiccup would crash the whole run on what looks like a perfectly reasonable prompt.
2. Validate. For each retrieved chunk, ask whether it actually helps answer the sub-question it was retrieved for:
def grade_relevance(sub_query, chunk_text):
prompt = (
f"Question: {sub_query}\n\nPassage: {chunk_text}\n\n"
"Does this passage help answer the question? Reply with only YES or NO."
)
verdict = generate(prompt).strip().upper()
return verdict.startswith("YES")
We’re using a simple binary relevance grader here for clarity. Production systems often use numeric relevance scores or cross-encoder rerankers instead.
3. Reformulate. If nothing came back relevant, rewrite the query instead of giving up:
def reformulate(sub_query):
prompt = (
f"This search query returned nothing useful: '{sub_query}'. "
"Rewrite it as a broader or differently-worded query that might "
"retrieve better results. Reply with only the rewritten query."
)
return generate(prompt).strip()
4. The orchestrator. This is what ties decompose → retrieve → validate → retry → generate into an actual loop. It also prints a trace of what it did at each step — for a system with this many moving pieces, “what actually happened” is worth seeing, not just trusting:
def agentic_rag(query, docs, max_retries=2, verbose=True):
sub_queries = decompose(query)
evidence = []
for i, sub_query in enumerate(sub_queries, start=1):
if verbose:
print(f"Sub-query {i}:")
attempt = 0
current_query = sub_query
found = False
while attempt <= max_retries and not found:
chunks = retrieve(current_query, docs, top_k=2)
if verbose:
print("Retrieve ->")
for chunk in chunks:
print(chunk["id"])
relevant_chunks = [
chunk for chunk in chunks if grade_relevance(sub_query, chunk["text"])
]
if verbose:
print("Validated:")
for chunk in relevant_chunks:
print(f"✓ {chunk['id']}")
if relevant_chunks:
evidence.extend(relevant_chunks)
found = True
else:
current_query = reformulate(current_query)
attempt += 1
if not evidence:
return "I couldn't find enough reliable information to answer that."
seen_ids = set()
unique_evidence = []
for chunk in evidence:
if chunk["id"] not in seen_ids:
unique_evidence.append(chunk)
seen_ids.add(chunk["id"])
if verbose:
print("Final evidence:")
for chunk in unique_evidence:
print(chunk["id"])
context = "\n\n".join(chunk["text"] for chunk in unique_evidence)
prompt = (
f"Answer the question using only the context below. If the context "
f"includes conflicting information, point out the conflict.\n\n"
f"Context:\n{context}\n\nQuestion: {query}"
)
return generate(prompt)
Run it against the test question below and you’ll get output shaped like this:
Sub-query 1:
Retrieve ->
rate_limits
errors
Validated:
✓ rate_limits
Sub-query 2:
Retrieve ->
pricing
auth
Validated:
✓ pricing
Final evidence:
rate_limits
pricing
Notice errors shows up under Retrieve -> but never gets a ✓ — it was fetched but didn't pass grade_relevance, so it's correctly absent from Final evidence. That's the trace doing its job: it shows the difference between what was fetched and what was actually used, which is exactly the gap a system with no validation step can't show you at all, because it never checks.
Notice what’s not here: no framework, no special “agent” class, no magic. It’s a loop with an escape hatch (max_retries), a print statement at each step, and a dedup step, calling the same four functions above in sequence. That's genuinely most of what "agentic RAG" means at the implementation level — everything else is refinement on top of this shape.
Running both side by side
if __name__ == "__main__":
indexed_docs = build_index(DOCS)
test_question = (
"I'm on the free plan and I got blocked after a bunch of requests "
"today. What's happening and what are my options?"
)
print("--- Traditional RAG ---")
print(traditional_rag(test_question, indexed_docs))
print("\n--- Agentic RAG ---")
print(agentic_rag(test_question, indexed_docs))
Run it yourself and compare the two outputs. The agentic_rag call will print its trace first, then the answer underneath it. What you should see: the traditional version embeds the whole question as one vector, which lands closest to rate_limits and errors — both are directly about being blocked — and its answer explains the 429 and tells you to wait or upgrade, but without the actual numbers, since pricing never gets pulled in. The agentic version decomposes the question, and one of its sub-queries ends up closer to something like "what tier limits and upgrade options are available," which retrieves pricing specifically — so its final answer includes the concrete numbers (100/day free, 10,000/day Pro) that the traditional version left out.
If you run this yourself and the two answers still come out equally complete, that’s worth noticing rather than ignoring — it usually means the embedding model found enough signal in rate_limits and errors combined to imply the tier structure even without the pricing doc, or that decompose didn't produce a sub-query specific enough to isolate pricing. Both are realistic failure modes, not bugs in the code — and they're exactly the kind of thing context-recall evaluation (covered in a later article in this series) is built to catch systematically instead of by eyeballing one run.
Part 2 — Rebuilding it with LlamaIndex and LangGraph
Same demo, same docs, same test question. This time, LlamaIndex handles indexing and retrieval, and LangGraph handles the loop as an actual state machine instead of a hand-rolled while loop.
pip install llama-index llama-index-llms-google-genai llama-index-embeddings-google-genai langgraph nest-asyncio
A heads-up if you’re running this in Jupyter: LlamaIndex’s GoogleGenAI integration calls asyncio.run() internally on every complete() call. Jupyter already runs its own event loop in the background, and Python won't let you nest a second asyncio.run() inside one that's already running. So without a fix, the decompose node will fail with RuntimeError: asyncio.run() cannot be called from a running event loop the moment you invoke the graph.
import nest_asyncio
nest_asyncio.apply()
Indexing — what LlamaIndex replaces
Compare this to the embed_text / build_index / cosine_similarity / retrieve block from Part 1:
import os
from llama_index.core import VectorStoreIndex, Document, Settings
from llama_index.embeddings.google_genai import GoogleGenAIEmbedding
from llama_index.llms.google_genai import GoogleGenAI
API_KEY = os.environ["GEMINI_API_KEY"]
Settings.embed_model = GoogleGenAIEmbedding(model_name="gemini-embedding-001", api_key=API_KEY)
Settings.llm = GoogleGenAI(model="gemini-3.6-flash", api_key=API_KEY)
DOCS = [
"ByteFlow API pricing tiers: Free tier costs $0 per month and includes 100 "
"requests per day, with no webhook access. Pro tier costs $29 per month and "
"includes 10,000 requests per day, plus webhook access. Enterprise tier has "
"custom volume pricing and is billed annually.",
"Rate limits are enforced per API key, not per account, and reset daily at "
"midnight UTC. Exceeding your allotted requests for the day returns an HTTP "
"429 response for the remainder of the current window. Your specific daily "
"request cap depends on your subscription tier - see the pricing page for "
"exact tier-by-tier numbers.",
"Error code reference: 401 means an invalid or missing API key. 429 means "
"you have exceeded your rate limit for the current window. 500 means an "
"internal server error, and requests should be retried with backoff. If "
"you're seeing repeated 429 responses, you can wait for your daily window "
"to reset, or move to a higher-volume subscription tier.",
"To authenticate requests, pass your API key in the Authorization header as "
"a Bearer token, e.g. 'Authorization: Bearer YOUR_API_KEY'. Keys are "
"generated from the dashboard under Settings > API Keys.",
"Webhooks let you receive real-time events when data changes, instead of "
"polling the API. Webhook access is only available on Pro tier and above. "
"Configure your endpoint URL from the dashboard under Settings > Webhooks.",
]
documents = [Document(text=t) for t in DOCS]
index = VectorStoreIndex.from_documents(documents)
retriever = index.as_retriever(similarity_top_k=2)
The VectorStoreIndex handles all chunking, embedding, and cosine-similarity search — the same steps Part 1 built by hand.
The loop as a state graph
This is where LangGraph earns its keep. Instead of a while loop with a manual retry counter, the loop becomes an explicit graph with nodes and conditional edges:
import json
from typing import TypedDict, List
from langgraph.graph import StateGraph, END
class AgentState(TypedDict):
query: str
sub_queries: List[str]
evidence: List[str]
retries: int
sufficient: bool
answer: str
def decompose_node(state: AgentState) -> AgentState:
prompt = (
"Break the following question into 1-3 self-contained sub-questions. "
"Respond with a JSON array of strings, nothing else.\n\n"
f"Question: {state['query']}"
)
raw = Settings.llm.complete(prompt).text
cleaned = raw.strip().strip("`").replace("json", "", 1).strip()
try:
sub_queries = json.loads(cleaned)
except json.JSONDecodeError:
sub_queries = [state["query"]]
return {**state, "sub_queries": sub_queries, "evidence": [], "retries": 0, "sufficient": False}
def retrieve_node(state: AgentState) -> AgentState:
new_evidence = list(state["evidence"])
for sub_query in state["sub_queries"]:
nodes = retriever.retrieve(sub_query)
new_evidence.extend(node.text for node in nodes)
return {**state, "evidence": new_evidence}
def validate_node(state: AgentState) -> AgentState:
prompt = (
f"Question: {state['query']}\n\nEvidence:\n"
f"{chr(10).join(state['evidence'])}\n\n"
"Is this evidence enough to answer the question fully? Reply with only YES or NO."
)
verdict = Settings.llm.complete(prompt).text.strip().upper()
sufficient = verdict.startswith("YES")
return {**state, "sufficient": sufficient, "retries": state["retries"] + (0 if sufficient else 1)}
def reformulate_node(state: AgentState) -> AgentState:
prompt = (
f"This search didn't return enough to answer: '{state['query']}'. "
"Rewrite it as a broader or differently-worded question that might "
"retrieve better results. Reply with only the rewritten question."
)
reformulated = Settings.llm.complete(prompt).text.strip()
return {**state, "query": reformulated, "sub_queries": [reformulated]}
def route_after_validate(state: AgentState) -> str:
if state["sufficient"] or state["retries"] > 2:
return "generate"
return "reformulate"
def generate_node(state: AgentState) -> AgentState:
context = "\n\n".join(dict.fromkeys(state["evidence"]))
prompt = (
f"Answer using only this context, and flag any conflicts:\n\n"
f"{context}\n\nQuestion: {state['query']}"
)
return {**state, "answer": Settings.llm.complete(prompt).text}
graph = StateGraph(AgentState)
graph.add_node("decompose", decompose_node)
graph.add_node("retrieve", retrieve_node)
graph.add_node("validate", validate_node)
graph.add_node("reformulate", reformulate_node)
graph.add_node("generate", generate_node)
graph.set_entry_point("decompose")
graph.add_edge("decompose", "retrieve")
graph.add_edge("retrieve", "validate")
graph.add_conditional_edges(
"validate", route_after_validate, {"reformulate": "reformulate", "generate": "generate"}
)
graph.add_edge("reformulate", "retrieve")
graph.add_edge("generate", END)
app = graph.compile()
Five nodes now, matching Part 1’s five pieces one for one: decompose, retrieve, validate, reformulate, generate. validate records a sufficient flag directly rather than making route_after_validate infer success from the retry count — that distinction matters: without a separate flag, a second attempt that succeeds would leave retries at 1, which is indistinguishable from "still failing," and the graph would loop forever instead of moving on. When validate finds the evidence insufficient, it routes to reformulate — which rewrites the query, exactly like Part 1's reformulate function — and reformulate loops back to retrieve. That loop repeats up to twice before route_after_validate gives up and moves on to generate anyway, same cap as Part 1's max_retries=2.
Notice that the control flow is now described declaratively. Instead of Python deciding what happens next line by line, the graph defines the possible transitions up front — which is what makes retries, branching, and checkpointing much easier to manage as the graph grows.
result = app.invoke({
"query": "I'm on the free plan and I got blocked after a bunch of requests today. What's happening and what are my options?",
"sub_queries": [], "evidence": [], "retries": 0, "sufficient": False, "answer": "",
})
print(result["answer"])Line count is actually a bit longer than Part 1’s version here — LangGraph’s overhead only starts paying off once the graph gets more complex than four nodes, or once you need the things it gives you for free: automatic state persistence between steps, the ability to checkpoint and resume a run that failed halfway through, and a graph structure you could visualize.
Visualize your graph
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))
Tradeoffs
- Visibility. In Part 1, every retry, every validation call, every prompt is a plain function you can print, breakpoint, or step through. In Part 2, a decent amount of that lives inside VectorStoreIndex and Settings.llm.complete — still inspectable, but one layer further away.
- Boilerplate. LlamaIndex cut the retrieval code by more than half — a real win the moment you go from 5 documents to 5,000, where you’d otherwise be hand-rolling chunking strategies and batched embedding calls.
- State handling. LangGraph’s AgentState and conditional edges are more code up front for a 4-node graph, but they stop being "more code" the moment you need to persist state across a crash, add a fifth node, or run this as part of a larger multi-agent system.
Real production agentic RAG systems layer in more — reranking, hybrid retrieval, tool routing, structured outputs, observability, caching, guardrails — but all of it builds on the same loop you just implemented twice: decompose, retrieve, validate, retry, generate. Everything else is refinement on top of that shape.