Decrease Hallucinations in a RAG Pipeline

Before building the configurable pipeline, run one query through a default RAG chain and see a real eval score. This is the hallucination the rest of the recipe hunts down.

from langchain_community.document_loaders.csv_loader import CSVLoader
from langchain_community.vectorstores import FAISS
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
from traceai_langchain import LangChainInstrumentor
from fi_instrumentation import register
from fi_instrumentation.fi_types import (
    EvalName,
    EvalSpanKind,
    EvalTag,
    EvalTagType,
    ProjectType,
)

trace_provider = register(
    project_type=ProjectType.EXPERIMENT,
    project_name="Experiment_RAG_Evaluation",
    project_version_name="baseline_stuff_chain",
    eval_tags=[
        EvalTag(
            type=EvalTagType.OBSERVATION_SPAN,
            value=EvalSpanKind.LLM,
            eval_name=EvalName.GROUNDEDNESS,
            config={},
            mapping={
                "input": "llm.input_messages.1.message.content",
                "output": "llm.output_messages.0.message.content",
            },
            custom_eval_name="Groundedness",
        ),
    ],
)
LangChainInstrumentor().instrument(tracer_provider=trace_provider)

documents = CSVLoader(file_path="./data.csv", encoding="utf-8").load()
chunks = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=150).split_documents(documents)
vectorstore = FAISS.from_documents(chunks, OpenAIEmbeddings(model="text-embedding-3-small"))
retriever = vectorstore.as_retriever(search_type="similarity", search_kwargs={"k": 3})
rag_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0.5),
    chain_type="stuff",
    retriever=retriever,
)

query = "Who were the three stars in the NHL game between Buffalo Sabres and Edmonton Oilers?"
result = rag_chain.invoke({"query": query})
print(result["result"])

You should see a printed answer, then in the Future AGI dashboard under Prototype → Experiment_RAG_Evaluation: a trace for the query with a Groundedness score. In an illustrative run this baseline scored 0.41, low, because the stuff chain packs every retrieved chunk into one prompt and the model filled a gap with a player name the context never mentioned. That drift is the hallucination the rest of this recipe works to reduce.

Put the pipeline’s tunable pieces in a config file so you can rerun the same experiment with a different chunking, retrieval, or chain strategy without touching code.

project/
├── data.csv             # question/context/answer rows to index and query
├── config.yaml          # experiment parameters
└── rag_experiment.py    # RAG setup and evaluation script

future_agi:
  project_name: "Experiment_RAG_Evaluation"
  project_version: "RecursiveCharacterTextSplitter_similarity_map_reduce"

openai:
  llm_model: "gpt-4o-mini"
  llm_temperature: 0.5
  embedding_model: "text-embedding-3-small"

data:
  file_path: "./data.csv"
  encoding: "utf-8"

chunking:
  enabled: true
  # Options: RecursiveCharacterTextSplitter, CharacterTextSplitter
  splitter_type: "RecursiveCharacterTextSplitter"
  chunk_size: 1000
  chunk_overlap: 150

retrieval:
  # Options: "similarity", "mmr" (Maximal Marginal Relevance)
  search_type: "similarity"
  k: 3

chain:
  # Options: "stuff", "map_reduce", "refine", "map_rerank"
  type: "map_reduce"
  return_source_documents: true

evaluation:
  queries:
    - "Who found the answer to a search query collar george herbert essay?"
    - "What are some of the potential negative impacts of charity as discussed in the context?"
    - "Who were the three stars in the NHL game between Buffalo Sabres and Edmonton Oilers?"

project_version becomes the run label in Future AGI. Give every configuration a distinct value so runs stay comparable in step 7.

register() and the OpenAI client pick up FI_API_KEY, FI_SECRET_KEY, FI_BASE_URL, and OPENAI_API_KEY from the environment variables you exported in Install. config.yaml only carries the experiment parameters, not credentials.

import yaml

def load_config(config_path: str) -> dict:
    try:
        with open(config_path, "r") as f:
            config = yaml.safe_load(f)
        print(f"Configuration loaded successfully from {config_path}")
        return config
    except FileNotFoundError:
        print(f"Error: Configuration file not found at {config_path}")
        exit(1)
    except yaml.YAMLError as e:
        print(f"Error parsing YAML file {config_path}: {e}")
        exit(1)

You should see:

Configuration loaded successfully from config.yaml

Three evals catch different failure modes: groundedness checks whether the answer is a well-supported, faithful response to the question, context adherence catches an answer that leaks in knowledge the retrieved context never provided, and context relevance catches a retriever that hands the model the wrong passages in the first place.

from traceai_langchain import LangChainInstrumentor
from fi_instrumentation import register
from fi_instrumentation.fi_types import (
    EvalName,
    EvalSpanKind,
    EvalTag,
    EvalTagType,
    ProjectType,
)

def setup_instrumentation(config: dict):
    eval_tags = [
        EvalTag(
            type=EvalTagType.OBSERVATION_SPAN,
            value=EvalSpanKind.LLM,
            eval_name=EvalName.GROUNDEDNESS,
            config={},
            mapping={
                "input": "llm.input_messages.1.message.content",
                "output": "llm.output_messages.0.message.content",
            },
            custom_eval_name="Groundedness",
        ),
        EvalTag(
            type=EvalTagType.OBSERVATION_SPAN,
            value=EvalSpanKind.LLM,
            eval_name=EvalName.CONTEXT_ADHERENCE,
            config={},
            mapping={
                "context": "llm.input_messages.0.message.content",
                "output": "llm.output_messages.0.message.content",
            },
            custom_eval_name="Context_Adherence",
        ),
        EvalTag(
            type=EvalTagType.OBSERVATION_SPAN,
            value=EvalSpanKind.LLM,
            eval_name=EvalName.CONTEXT_RELEVANCE,
            config={"check_internet": False},
            mapping={
                "input": "llm.input_messages.1.message.content",
                "context": "llm.input_messages.0.message.content",
            },
            custom_eval_name="Context_Relevance",
        ),
    ]

    trace_provider = register(
        project_type=ProjectType.EXPERIMENT,
        project_name=config["future_agi"]["project_name"],
        project_version_name=config["future_agi"]["project_version"],
        eval_tags=eval_tags,
    )
    LangChainInstrumentor().instrument(tracer_provider=trace_provider)
    print(f"Instrumentation ready for project: {config['future_agi']['project_name']}")

CONTEXT_RELEVANCE takes context and input only, no output: it scores whether the retriever pulled passages relevant to the question, before the model ever answers. check_internet controls whether the eval is allowed to verify claims against a live web search; leave it False for a pipeline that should be judged on its own retrieved context.

You should see:

Instrumentation ready for project: Experiment_RAG_Evaluation

Chunk the documents, embed them, index them in FAISS, and wrap the retriever in a RetrievalQA chain using the strategies named in config.yaml.

from langchain_community.document_loaders.csv_loader import CSVLoader
from langchain_community.vectorstores import FAISS
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter, RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA

def setup_rag(config: dict):
    data_config = config["data"]
    chunking_config = config["chunking"]
    retrieval_config = config["retrieval"]
    chain_config = config["chain"]
    openai_config = config["openai"]

    loader = CSVLoader(file_path=data_config["file_path"], encoding=data_config["encoding"])
    documents = loader.load()
    print(f"Loaded {len(documents)} documents.")

    if chunking_config["enabled"]:
        splitter_cls = (
            RecursiveCharacterTextSplitter
            if chunking_config["splitter_type"] == "RecursiveCharacterTextSplitter"
            else CharacterTextSplitter
        )
        text_splitter = splitter_cls(
            chunk_size=chunking_config["chunk_size"],
            chunk_overlap=chunking_config["chunk_overlap"],
        )
        docs_to_index = text_splitter.split_documents(documents)
        print(f"Split into {len(docs_to_index)} chunks.")
    else:
        docs_to_index = documents

    embeddings = OpenAIEmbeddings(model=openai_config["embedding_model"])
    vectorstore = FAISS.from_documents(docs_to_index, embeddings)

    retriever_kwargs = {"k": retrieval_config["k"]}
    if retrieval_config["search_type"] == "mmr":
        retriever_kwargs["fetch_k"] = retrieval_config.get("fetch_k", 20)
        retriever_kwargs["lambda_mult"] = retrieval_config.get("lambda_mult", 0.5)

    retriever = vectorstore.as_retriever(
        search_type=retrieval_config["search_type"],
        search_kwargs=retriever_kwargs,
    )

    llm = ChatOpenAI(temperature=openai_config["llm_temperature"], model=openai_config["llm_model"])

    rag_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type=chain_config["type"],
        retriever=retriever,
        return_source_documents=chain_config["return_source_documents"],
    )
    print("RAG chain setup complete.")
    return rag_chain

You should see:

Loaded <N> documents.
Split into <N> chunks.
RAG chain setup complete.

def process_query(rag_chain, query: str):
    result = rag_chain.invoke({"query": query})
    return result.get("result", "No answer could be generated.")


def run_evaluation_queries(config: dict):
    rag_chain = setup_rag(config)
    results = {}
    for query in config["evaluation"]["queries"]:
        response = process_query(rag_chain, query)
        print(f"Q: {query}\nA: {response}\n")
        results[query] = response
    print(f"Project: {config['future_agi']['project_name']}, Version: {config['future_agi']['project_version']}")
    return results


if __name__ == "__main__":
    config = load_config("config.yaml")
    setup_instrumentation(config)
    run_evaluation_queries(config)

Each rag_chain.invoke() call is captured as a trace, with the three eval tags scoring the LLM span automatically. No separate evaluate() call is needed: the scores land in Future AGI as the trace lands.

You should see the script print an answer per query, then in the Future AGI dashboard: Prototype → your project → a new run under Experiment_RAG_Evaluation with a trace per query and Groundedness, Context Adherence, and Context Relevance scores on each.

Edit config.yaml, give project_version a new name (for example CharacterTextSplitter_mmr_map_rerank), and rerun the script. Repeat for the chunking, retrieval, and chain combinations you want to compare.

Open All Runs for the project, switch to the Summary tab, and click Choose Winner (crown icon) to weight groundedness, context adherence, and context relevance against cost and latency.

Weights are set per project and persist across every rerun you trigger afterward

The baseline stuff-chain run from step 1 sorts to the bottom of this table

In an illustrative run with these weights, CharacterTextSplitter_mmr_map_rerank ranked highest: character-based chunking, MMR retrieval, and a map-rerank chain, scoring 0.89 groundedness, 0.92 context adherence, and 0.95 context relevance against the baseline_stuff_chain run’s 0.41 groundedness from step 1, a 0.41 → 0.89 improvement. Your own ranking depends on your data and queries. Run the comparison to find yours.