Generate TestSet Questions to evaluate your RAG pipeline¶

Tech Stack: LangGraph, Qdrant, OpenAI and TruLens

TruLens benchmark to generate the questions

Installation¶

In [ ]:

!pip install langgraph langchain-community langchain-openai langchain-qdrant pypdfium2
!pip install trulens "trulens-benchmark"
!pip install trulens-providers-openai
!pip install trulens-apps-langgraph

!pip install langgraph langchain-community langchain-openai langchain-qdrant pypdfium2 !pip install trulens "trulens-benchmark" !pip install trulens-providers-openai !pip install trulens-apps-langgraph

Initial Setup¶

Install the required libraries and configure API keys for OpenAI and Qdrant. A TruLens session is initialized here to track and store all evaluation data throughout the notebook.

• Get OpenAI API Key: https://platform.openai.com/
• Get Qdrant API Key and Endpoint URL: https://cloud.qdrant.io/

In [ ]:

import os

from google.colab import userdata

os.environ["OPENAI_API_KEY"] = userdata.get("OPENAI_API_KEY")
QDRANT_URL = userdata.get("QDRANT_URL")
QDRANT_API_KEY = userdata.get("QDRANT_API_KEY")

import os from google.colab import userdata os.environ["OPENAI_API_KEY"] = userdata.get("OPENAI_API_KEY") QDRANT_URL = userdata.get("QDRANT_URL") QDRANT_API_KEY = userdata.get("QDRANT_API_KEY")

Overview¶

We will generate a dataset to evaluate the RAG pipeline built using LangGraph and Qdrant, and assess it using the RAG Triad evaluation metrics.

In [ ]:

from langchain_community.document_loaders import PyPDFium2Loader
from langchain_openai import OpenAIEmbeddings
from langchain_qdrant import QdrantVectorStore
from langchain_text_splitters import RecursiveCharacterTextSplitter
from qdrant_client import QdrantClient
from qdrant_client import models

from langchain_community.document_loaders import PyPDFium2Loader from langchain_openai import OpenAIEmbeddings from langchain_qdrant import QdrantVectorStore from langchain_text_splitters import RecursiveCharacterTextSplitter from qdrant_client import QdrantClient from qdrant_client import models

In [ ]:

from trulens.core import TruSession

session = TruSession()
session.reset_database()

from trulens.core import TruSession session = TruSession() session.reset_database()

Data loading and chunking¶

The Economic Survey 2025-26 report is downloaded and parsed using PyPDFium2. It is then split into overlapping chunks using Langchain's RecursiveCharacterTextSplitter to prepare the text for embedding and retrieval.

Economic Survey 2025-26 by Government of India: Survey Report

In [ ]:

!curl -L -A "Mozilla/5.0" https://www.indiabudget.gov.in/economicsurvey/doc/eschapter/epreface.pdf -o survey.pdf

!curl -L -A "Mozilla/5.0" https://www.indiabudget.gov.in/economicsurvey/doc/eschapter/epreface.pdf -o survey.pdf

In [ ]:

path = "survey.pdf"
loader = PyPDFium2Loader(path)
docs = loader.load()

path = "survey.pdf" loader = PyPDFium2Loader(path) docs = loader.load()

In [ ]:

print(len(docs))

print(len(docs))

In [ ]:

print(docs[1].page_content)

print(docs[1].page_content)

In [ ]:

print(docs[1].metadata)

print(docs[1].metadata)

In [ ]:

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1536,
    chunk_overlap=0,
)
chunks = text_splitter.split_documents(docs)

text_splitter = RecursiveCharacterTextSplitter( chunk_size=1536, chunk_overlap=0, ) chunks = text_splitter.split_documents(docs)

Setup your Qdrant Vector Database¶

A Qdrant collection is configured using dense vectors generated from OpenAI embeddings, enabling semantic search based on meaning and contextual similarity. Each document or chunk is converted into its corresponding embedding and indexed within the collection along with its associated metadata.

During retrieval, the query is embedded into the same vector space and matched against the indexed vectors to return the most semantically relevant results.

In [ ]:

embeddings_model = OpenAIEmbeddings(model="text-embedding-3-small")

embeddings_model = OpenAIEmbeddings(model="text-embedding-3-small")

In [ ]:

print(len(embeddings_model.embed_query("this is testing to check dimensions")))

print(len(embeddings_model.embed_query("this is testing to check dimensions")))

In [ ]:

collection_name = "economicv1"

collection_name = "economicv1"

In [ ]:

client = QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY)

client = QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY)

In [ ]:

client.create_collection(
    collection_name=collection_name,
    vectors_config={
        "dense": models.VectorParams(size=1536, distance=models.Distance.COSINE)
    },
)

client.create_collection( collection_name=collection_name, vectors_config={ "dense": models.VectorParams(size=1536, distance=models.Distance.COSINE) }, )

In [ ]:

db = QdrantVectorStore(
    client=client,
    collection_name=collection_name,
    embedding=embeddings_model,
    vector_name="dense",
)

db = QdrantVectorStore( client=client, collection_name=collection_name, embedding=embeddings_model, vector_name="dense", )

Test the retriver¶

In [ ]:

db.add_documents(
    documents=chunks
)  # just run it once. After the data is indexed, you need not have to add_documents again

db.add_documents( documents=chunks ) # just run it once. After the data is indexed, you need not have to add_documents again

In [ ]:

query = "what is Financial Time wrote on the eve of Christmas 2025"
relevant_docs = db.max_marginal_relevance_search(query)

query = "what is Financial Time wrote on the eve of Christmas 2025" relevant_docs = db.max_marginal_relevance_search(query)

In [ ]:

relevant_docs[0].page_content

relevant_docs[0].page_content

Convert to the retriever: i.e., Langchain object

In [ ]:

retriever = db.as_retriever()

retriever = db.as_retriever()

In [ ]:

print(retriever.invoke(query)[0].page_content)

print(retriever.invoke(query)[0].page_content)

Build your LangGraph RAG workflow - Answer Generation¶

In [ ]:

from langchain_openai import ChatOpenAI

from langchain_openai import ChatOpenAI

In [ ]:

llm = ChatOpenAI(model_name="gpt-5-mini")

llm = ChatOpenAI(model_name="gpt-5-mini")

In [ ]:

SYSTEM_PROMPT = """
  You are an expert financial analyst specializing in corporate finance.
  You should be respectful and truthful while answering the user questions, if not you will face serious consequences.

  The only source information you have is the context provided, if the user query is not from the context
  Just say `I dont know , not enough information provided.`
"""

SYSTEM_PROMPT = """ You are an expert financial analyst specializing in corporate finance. You should be respectful and truthful while answering the user questions, if not you will face serious consequences. The only source information you have is the context provided, if the user query is not from the context Just say `I dont know , not enough information provided.` """

In [ ]:

USER_PROMPT = """
  Answer the USER QUERY based on the CONTEXT below.
  If the question cannot be answered using the information provided answer with `I dont know , not enough information provided.`

  <context>
  CONTEXT: {context}
  </context>

  <query>
  USER QUERY: {query}
  </query>
"""

USER_PROMPT = """ Answer the USER QUERY based on the CONTEXT below. If the question cannot be answered using the information provided answer with `I dont know , not enough information provided.` <context> CONTEXT: {context} </context> <query> USER QUERY: {query} </query> """

In [ ]:

from typing import List, TypedDict

from langgraph.graph import END
from langgraph.graph import START
from langgraph.graph import StateGraph  # node and edges logic
from trulens.core.otel.instrument import instrument
from trulens.otel.semconv.trace import SpanAttributes

from typing import List, TypedDict from langgraph.graph import END from langgraph.graph import START from langgraph.graph import StateGraph # node and edges logic from trulens.core.otel.instrument import instrument from trulens.otel.semconv.trace import SpanAttributes

In [ ]:

class RAGState(TypedDict):
    query: str
    context: List[str]
    answer: str

class RAGState(TypedDict): query: str context: List[str] answer: str

In [ ]:

@instrument(
    span_type=SpanAttributes.SpanType.RETRIEVAL,
    attributes={
        SpanAttributes.RETRIEVAL.QUERY_TEXT: "query",
        SpanAttributes.RETRIEVAL.RETRIEVED_CONTEXTS: "return",
    },
)
def retrieve_docs(query: str) -> list:
    results = db.max_marginal_relevance_search(query=query, k=4)
    return [result.page_content for result in results]

@instrument( span_type=SpanAttributes.SpanType.RETRIEVAL, attributes={ SpanAttributes.RETRIEVAL.QUERY_TEXT: "query", SpanAttributes.RETRIEVAL.RETRIEVED_CONTEXTS: "return", }, ) def retrieve_docs(query: str) -> list: results = db.max_marginal_relevance_search(query=query, k=4) return [result.page_content for result in results]

In [ ]:

@instrument(span_type=SpanAttributes.SpanType.GENERATION)
def generate_answer(context: str, query: str) -> str:
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {
            "role": "user",
            "content": USER_PROMPT.format(context=context, query=query),
        },
    ]
    response = llm.invoke(messages)
    return response.content

@instrument(span_type=SpanAttributes.SpanType.GENERATION) def generate_answer(context: str, query: str) -> str: messages = [ {"role": "system", "content": SYSTEM_PROMPT}, { "role": "user", "content": USER_PROMPT.format(context=context, query=query), }, ] response = llm.invoke(messages) return response.content

In [ ]:

@instrument(
    span_type=SpanAttributes.SpanType.RECORD_ROOT,
    attributes={
        SpanAttributes.RECORD_ROOT.INPUT: "question",
        SpanAttributes.RECORD_ROOT.OUTPUT: "return",
    },
)
def execute_graph(question: str) -> str:
    result = graph.invoke({"query": question})
    return result["answer"]

@instrument( span_type=SpanAttributes.SpanType.RECORD_ROOT, attributes={ SpanAttributes.RECORD_ROOT.INPUT: "question", SpanAttributes.RECORD_ROOT.OUTPUT: "return", }, ) def execute_graph(question: str) -> str: result = graph.invoke({"query": question}) return result["answer"]

In [ ]:

def search(state: RAGState) -> RAGState:
    state["context"] = retrieve_docs(state["query"])
    return state


def answer(state: RAGState) -> RAGState:
    context = "\n".join(state["context"])
    state["answer"] = generate_answer(context, state["query"])
    return state

def search(state: RAGState) -> RAGState: state["context"] = retrieve_docs(state["query"]) return state def answer(state: RAGState) -> RAGState: context = "\n".join(state["context"]) state["answer"] = generate_answer(context, state["query"]) return state

In [ ]:

# Agent - workflows
workflow = StateGraph(RAGState)
workflow.add_node("search_context", search)
workflow.add_node("answer_generation", answer)

workflow.add_edge(START, "search_context")
workflow.add_edge("search_context", "answer_generation")
workflow.add_edge("answer_generation", END)

graph = workflow.compile()

# Agent - workflows workflow = StateGraph(RAGState) workflow.add_node("search_context", search) workflow.add_node("answer_generation", answer) workflow.add_edge(START, "search_context") workflow.add_edge("search_context", "answer_generation") workflow.add_edge("answer_generation", END) graph = workflow.compile()

In [ ]:

graph

graph

In [ ]:

print(execute_graph(query))

print(execute_graph(query))

Generate Testset using TruLens¶

TruLens allows you to generate a test set of a specified breadth and depth, tailored to your app and data. The resulting test set will be a list of test prompts of length depth, for breadth categories of prompts. This test set will be made up of breadth X depth prompts organized by prompt category.

In [ ]:

from trulens.benchmark.generate.generate_test_set import GenerateTestSet
from trulens.providers.openai import OpenAI

from trulens.benchmark.generate.generate_test_set import GenerateTestSet from trulens.providers.openai import OpenAI

In [ ]:

test = GenerateTestSet(app_callable=execute_graph)
provider = OpenAI()

test = GenerateTestSet(app_callable=execute_graph) provider = OpenAI()

In [ ]:

test_set = test.generate_test_set(
    test_breadth=3,
    test_depth=2,
)

# breadth = 3 → generate 3 categories
# depth = 2 → generate 2 questions

test_set = test.generate_test_set( test_breadth=3, test_depth=2, ) # breadth = 3 → generate 3 categories # depth = 2 → generate 2 questions

In [ ]:

test_set

test_set

Run Evals¶

Evaluate with TruLens feedbacks

In [ ]:

import numpy as np
from trulens.apps.langgraph import TruGraph
from trulens.core import Metric
from trulens.core import Selector

import numpy as np from trulens.apps.langgraph import TruGraph from trulens.core import Metric from trulens.core import Selector

In [ ]:

f_answer_relevance = Metric(
    implementation=provider.relevance_with_cot_reasons,
    name="Answer Relevance",
    selectors={
        "prompt": Selector.select_record_input(),
        "response": Selector.select_record_output(),
    },
)

f_context_relevance = Metric(
    implementation=provider.context_relevance_with_cot_reasons,
    name="Context Relevance",
    selectors={
        "question": Selector.select_record_input(),
        "context": Selector(
            span_type=SpanAttributes.SpanType.RETRIEVAL,
            span_attribute=SpanAttributes.RETRIEVAL.RETRIEVED_CONTEXTS,
            collect_list=False,
        ),
    },
    agg=np.mean,
)

f_groundedness = Metric(
    implementation=provider.groundedness_measure_with_cot_reasons,
    name="Groundedness",
    selectors={
        "source": Selector(
            span_type=SpanAttributes.SpanType.RETRIEVAL,
            span_attribute=SpanAttributes.RETRIEVAL.RETRIEVED_CONTEXTS,
            collect_list=True,
        ),
        "statement": Selector.select_record_output(),
    },
)

f_answer_relevance = Metric( implementation=provider.relevance_with_cot_reasons, name="Answer Relevance", selectors={ "prompt": Selector.select_record_input(), "response": Selector.select_record_output(), }, ) f_context_relevance = Metric( implementation=provider.context_relevance_with_cot_reasons, name="Context Relevance", selectors={ "question": Selector.select_record_input(), "context": Selector( span_type=SpanAttributes.SpanType.RETRIEVAL, span_attribute=SpanAttributes.RETRIEVAL.RETRIEVED_CONTEXTS, collect_list=False, ), }, agg=np.mean, ) f_groundedness = Metric( implementation=provider.groundedness_measure_with_cot_reasons, name="Groundedness", selectors={ "source": Selector( span_type=SpanAttributes.SpanType.RETRIEVAL, span_attribute=SpanAttributes.RETRIEVAL.RETRIEVED_CONTEXTS, collect_list=True, ), "statement": Selector.select_record_output(), }, )

In [ ]:

tru_recorder = TruGraph(
    execute_graph,
    app_name="evals_result",
    app_version="v1",
    feedbacks=[f_groundedness, f_answer_relevance, f_context_relevance],
)

tru_recorder = TruGraph( execute_graph, app_name="evals_result", app_version="v1", feedbacks=[f_groundedness, f_answer_relevance, f_context_relevance], )

In [ ]:

with tru_recorder as recording:
    for topic, questions in test_set.items():
        for question in questions:
            execute_graph(question)

with tru_recorder as recording: for topic, questions in test_set.items(): for question in questions: execute_graph(question)

In [ ]:

session.get_leaderboard()

session.get_leaderboard()

In [ ]:

# Check what spans are actually being recorded
records = session.get_records_and_feedback()[0]
records

# Check what spans are actually being recorded records = session.get_records_and_feedback()[0] records