Skip to content

RAGTitan and factory

RAGTitan: the easiest way to use the package.

It builds every part for you, so you only need two calls:

from rag_titan import RAGTitan

rag = RAGTitan(rag_type="simple")
rag.index(["handbook.pdf", "policy.docx"])
print(rag.ask("How many leave days do I get?"))

print(rag.run("How many leave days do I get?")["log"])   # every step
rag.evaluate("questions.json")                           # end to end, uses the LLM
rag.evaluate_retrieval("questions.json")                 # search only, fast
rag.evaluate_stages(files, "golden.json")                # every stage, scorecard

RAGTitan

One object that can index files and answer questions about them.

Parameters:

Name Type Description Default
rag_type

which RAG to use, "simple" or "adaptive".

 'simple'
settings

optional dict of settings to change, for example {"llm_model": "llama3.2"}. Anything missing uses the value from the environment or the default.

 None
Source code in src/rag_titan/app.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
class RAGTitan:
    """
    One object that can index files and answer questions about them.

    Args:
        rag_type: which RAG to use, "simple" or "adaptive".
        settings: optional dict of settings to change, for example
            {"llm_model": "llama3.2"}. Anything missing uses the
            value from the environment or the default.
    """

    def __init__(self, rag_type="simple", settings=None):
        self.settings = get_settings(settings)
        self.rag_type = rag_type

        # build all the parts once, they share one embedder and one store
        self.parts = build_rag_parts(self.settings)
        self.pipeline = self.parts["pipeline"]
        self.rag = build_rag(rag_type, self.parts)

    def index(self, file_paths, reset=True):
        """
        Read, chunk, embed and store the given files.

        Args:
            file_paths: list of paths to .pdf or .docx files.
            reset: when True, remove previously indexed chunks first.

        Returns:
            The number of chunks stored.
        """
        return self.pipeline.run(file_paths, reset=reset)

    def ask(self, question):
        """
        Answer a question using the indexed files.

        Args:
            question: the user's question as text.

        Returns:
            The answer text.
        """
        return self.rag.answer(question)

    def run(self, question):
        """
        Answer a question and show every step taken.

        Returns:
            The dict from the RAG type's run() ("answer", "sources", "steps"),
            plus "log": the steps as readable text (see describe_steps).
        """
        result = self.rag.run(question)
        result["log"] = describe_steps(result["steps"])
        return result

    def evaluate(self, questions, show=True):
        """
        Answer every test question with this RAG type and score the runs (uses the LLM).

        Args:
            questions: path to a questions JSON file, or a list of question dicts.
            show: print the report table.

        Returns:
            The dict from evaluate_rag(): evidence scores on the chunks given to
            the LLM, answered share, steps, time, and one row per question.
        """
        result = evaluate_rag(self.rag, questions, name=self.rag_type)
        if show:
            print(format_rag_report([result]))
        return result

    def evaluate_retrieval(self, questions, top_k=5, show=True):
        """
        Check how well search finds the evidence for test questions (no LLM used).

        Args:
            questions: path to a questions JSON file, or a list of question dicts.
            top_k: how many chunks to retrieve per question.
            show: print the report table.

        Returns:
            The dict from evaluate_retriever(): means, latency and one row per question.
        """
        result = evaluate_retriever(self.parts["retriever"], as_questions(questions), top_k)
        if show:
            print(format_report([result]))
        return result

    def evaluate_stages(self, file_paths, questions, k=5, include_embeddings=False, show=True):
        """
        Index the files, answer the golden questions and score every stage (uses the LLM).

        Args:
            file_paths: list of files to index (the index is rebuilt).
            questions: path to a golden questions JSON file, or a list of dicts
                with "id", "question" and gold labels (see collect_dataset).
            k: how many results to score in the ranking metrics.
            include_embeddings: also score the embedding stage.
            show: print the overall score, status and first failing stage.

        Returns:
            The report from RAGEvaluator: one entry per stage, "root_causes"
            and "scorecard". Save it with ReportManager.
        """
        dataset = collect_dataset(self, file_paths, questions, include_embeddings)
        report = RAGEvaluator(k=k).evaluate(**dataset)
        if show:
            print(summary_text(report))
        return report

ask(question)

Answer a question using the indexed files.

Parameters:

Name Type Description Default
question

the user's question as text.

 required

Returns:

Type Description

The answer text.
Source code in src/rag_titan/app.py 
62
63
64
65
66
67
68
69
70
71
72
def ask(self, question):
    """
    Answer a question using the indexed files.

    Args:
        question: the user's question as text.

    Returns:
        The answer text.
    """
    return self.rag.answer(question)

evaluate(questions, show=True)

Answer every test question with this RAG type and score the runs (uses the LLM).

Parameters:

Name Type Description Default
questions

path to a questions JSON file, or a list of question dicts.

 required
show

print the report table.

 True

Returns:

Type Description

The dict from evaluate_rag(): evidence scores on the chunks given to

the LLM, answered share, steps, time, and one row per question.
Source code in src/rag_titan/app.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def evaluate(self, questions, show=True):
    """
    Answer every test question with this RAG type and score the runs (uses the LLM).

    Args:
        questions: path to a questions JSON file, or a list of question dicts.
        show: print the report table.

    Returns:
        The dict from evaluate_rag(): evidence scores on the chunks given to
        the LLM, answered share, steps, time, and one row per question.
    """
    result = evaluate_rag(self.rag, questions, name=self.rag_type)
    if show:
        print(format_rag_report([result]))
    return result

evaluate_retrieval(questions, top_k=5, show=True)

Check how well search finds the evidence for test questions (no LLM used).

Parameters:

Name Type Description Default
questions

path to a questions JSON file, or a list of question dicts.

 required
top_k

how many chunks to retrieve per question.

 5
show

print the report table.

 True

Returns:

Type Description

The dict from evaluate_retriever(): means, latency and one row per question.
Source code in src/rag_titan/app.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def evaluate_retrieval(self, questions, top_k=5, show=True):
    """
    Check how well search finds the evidence for test questions (no LLM used).

    Args:
        questions: path to a questions JSON file, or a list of question dicts.
        top_k: how many chunks to retrieve per question.
        show: print the report table.

    Returns:
        The dict from evaluate_retriever(): means, latency and one row per question.
    """
    result = evaluate_retriever(self.parts["retriever"], as_questions(questions), top_k)
    if show:
        print(format_report([result]))
    return result

evaluate_stages(file_paths, questions, k=5, include_embeddings=False, show=True)

Index the files, answer the golden questions and score every stage (uses the LLM).

Parameters:

Name Type Description Default
file_paths

list of files to index (the index is rebuilt).

 required
questions

path to a golden questions JSON file, or a list of dicts with "id", "question" and gold labels (see collect_dataset).

 required
k

how many results to score in the ranking metrics.

 5
include_embeddings

also score the embedding stage.

 False
show

print the overall score, status and first failing stage.

 True

Returns:

Type Description

The report from RAGEvaluator: one entry per stage, "root_causes"

and "scorecard". Save it with ReportManager.
Source code in src/rag_titan/app.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def evaluate_stages(self, file_paths, questions, k=5, include_embeddings=False, show=True):
    """
    Index the files, answer the golden questions and score every stage (uses the LLM).

    Args:
        file_paths: list of files to index (the index is rebuilt).
        questions: path to a golden questions JSON file, or a list of dicts
            with "id", "question" and gold labels (see collect_dataset).
        k: how many results to score in the ranking metrics.
        include_embeddings: also score the embedding stage.
        show: print the overall score, status and first failing stage.

    Returns:
        The report from RAGEvaluator: one entry per stage, "root_causes"
        and "scorecard". Save it with ReportManager.
    """
    dataset = collect_dataset(self, file_paths, questions, include_embeddings)
    report = RAGEvaluator(k=k).evaluate(**dataset)
    if show:
        print(summary_text(report))
    return report

index(file_paths, reset=True)

Read, chunk, embed and store the given files.

Parameters:

Name Type Description Default
file_paths

list of paths to .pdf or .docx files.

 required
reset

when True, remove previously indexed chunks first.

 True

Returns:

Type Description

The number of chunks stored.
Source code in src/rag_titan/app.py 
49
50
51
52
53
54
55
56
57
58
59
60
def index(self, file_paths, reset=True):
    """
    Read, chunk, embed and store the given files.

    Args:
        file_paths: list of paths to .pdf or .docx files.
        reset: when True, remove previously indexed chunks first.

    Returns:
        The number of chunks stored.
    """
    return self.pipeline.run(file_paths, reset=reset)

run(question)

Answer a question and show every step taken.

Returns:

Type Description

The dict from the RAG type's run() ("answer", "sources", "steps"),

plus "log": the steps as readable text (see describe_steps).
Source code in src/rag_titan/app.py 
74
75
76
77
78
79
80
81
82
83
84
def run(self, question):
    """
    Answer a question and show every step taken.

    Returns:
        The dict from the RAG type's run() ("answer", "sources", "steps"),
        plus "log": the steps as readable text (see describe_steps).
    """
    result = self.rag.run(question)
    result["log"] = describe_steps(result["steps"])
    return result

The only place that knows which class to use for each setting.

Adding a new provider or strategy
  1. create its class in the right folder
  2. add one if name == ... line in the matching build_* function
  3. change the setting in .env (or pass it in get_settings())

build_agent_llm(settings)

Create the chat model used by agents (it must support tools).

Source code in src/rag_titan/factory.py 
52
53
54
def build_agent_llm(settings):
    """Create the chat model used by agents (it must support tools)."""
    return build_llm(settings, model=settings["agent_model"], think=settings["agent_think"])

build_chunker(settings)

Create the chunker named in settings["chunker"]. chunk_size and chunk_overlap count words, characters or tokens, depending on the chunker.

Source code in src/rag_titan/factory.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def build_chunker(settings):
    """
    Create the chunker named in settings["chunker"].
    chunk_size and chunk_overlap count words, characters or tokens,
    depending on the chunker.
    """
    name = settings["chunker"]
    size, overlap = settings["chunk_size"], settings["chunk_overlap"]

    if name == "word":
        return WordChunker(size, overlap)

    if name == "character":
        return CharacterChunker(size, overlap)

    if name == "token":
        return TokenChunker(size, overlap, encoding=settings["chunk_encoding"])

    if name == "recursive":
        return RecursiveChunker(size, overlap)

    # structure-aware chunkers: whole sentences, paragraphs, sections or pages
    structure_aware = {
        "sentence": SentenceChunker,
        "paragraph": ParagraphChunker,
        "heading": HeadingChunker,
        "page": PageChunker,
    }
    if name in structure_aware:
        return structure_aware[name](size, overlap)

    raise ValueError(f"Unknown chunker: {name}")

build_embedder(settings)

Create the embedding model named in settings["embed_provider"].

Source code in src/rag_titan/factory.py 
57
58
59
60
61
62
63
64
65
66
67
def build_embedder(settings):
    """Create the embedding model named in settings["embed_provider"]."""
    name = settings["embed_provider"]

    if name == "ollama":
        return OllamaEmbedder(settings["embed_model"], host=settings["ollama_host"])

    if name == "hash":
        return HashEmbedder()

    raise ValueError(f"Unknown embedding provider: {name}")

build_generator(settings, llm)

Create the answer generator named in settings["generator"].

Source code in src/rag_titan/factory.py 
130
131
132
133
134
135
136
137
def build_generator(settings, llm):
    """Create the answer generator named in settings["generator"]."""
    name = settings["generator"]

    if name == "grounded":
        return GroundedGenerator(llm)

    raise ValueError(f"Unknown generator: {name}")

build_llm(settings, model=None, think=False)

Create the chat model named in settings["llm_provider"].

Source code in src/rag_titan/factory.py 
41
42
43
44
45
46
47
48
49
def build_llm(settings, model=None, think=False):
    """Create the chat model named in settings["llm_provider"]."""
    name = settings["llm_provider"]
    model = model or settings["llm_model"]

    if name == "ollama":
        return OllamaLLM(model, host=settings["ollama_host"], think=think)

    raise ValueError(f"Unknown LLM provider: {name}")

build_rag(rag_type, parts)

Create one RAG type from the parts made by build_rag_parts().

Parameters:

Name Type Description Default
rag_type

one of RAG_TYPES, for example "simple" or "adaptive".

 required
parts

the dict returned by build_rag_parts().

 required
Source code in src/rag_titan/factory.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def build_rag(rag_type, parts):
    """
    Create one RAG type from the parts made by build_rag_parts().

    Args:
        rag_type: one of RAG_TYPES, for example "simple" or "adaptive".
        parts: the dict returned by build_rag_parts().
    """
    if rag_type == "simple":
        return SimpleRAG(parts["retriever"], parts["reranker"], parts["generator"])

    if rag_type == "adaptive":
        return AdaptiveRAG(
            parts["llm"], parts["retriever"], parts["reranker"], parts["generator"]
        )

    raise ValueError(f"Unknown RAG type: {rag_type}. Choose from {RAG_TYPES}")

build_rag_parts(settings)

Build every RAG building block, sharing one embedder and one store.

Returns:

Type Description

A dict with "llm", "pipeline", "retriever", "reranker"

and "generator".
Source code in src/rag_titan/factory.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def build_rag_parts(settings):
    """
    Build every RAG building block, sharing one embedder and one store.

    Returns:
        A dict with "llm", "pipeline", "retriever", "reranker"
        and "generator".
    """
    llm = build_llm(settings)
    embedder = build_embedder(settings)
    store = build_store(settings)

    return {
        "llm": llm,
        "pipeline": IndexingPipeline(build_chunker(settings), embedder, store),
        "retriever": DenseRetriever(embedder, store, dedupe_overlap=settings["dedupe_overlap"]),
        "reranker": build_reranker(settings, llm),
        "generator": build_generator(settings, llm),
    }

build_reranker(settings, llm=None)

Create the reranker named in settings["reranker"].

llm is not used yet. It is here for future LLM-based rerankers.

Source code in src/rag_titan/factory.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def build_reranker(settings, llm=None):
    """
    Create the reranker named in settings["reranker"].

    `llm` is not used yet. It is here for future LLM-based rerankers.
    """
    name = settings["reranker"]

    if name == "keyword":
        return KeywordReranker()
    if name == "none":
        return NoReranker()

    raise ValueError(f"Unknown reranker: {name}")

build_store(settings)

Create the vector store named in settings["vector_store"].

Source code in src/rag_titan/factory.py 
104
105
106
107
108
109
110
111
def build_store(settings):
    """Create the vector store named in settings["vector_store"]."""
    name = settings["vector_store"]

    if name == "memory":
        return MemoryStore()

    raise ValueError(f"Unknown vector store: {name}")

All settings in one place.

Values are read from environment variables (or a .env file in the folder where you run your code). Every setting has a safe default, so the package works without any configuration.

Example

from rag_titan.config import get_settings

settings = get_settings({"llm_model": "llama3.2", "chunk_size": 200})

get_settings(overrides=None)

Build the settings dictionary.

Parameters:

Name Type Description Default
overrides

optional dict of settings that win over the environment, for example {"chunk_size": 200}.

 None

Returns:

Type Description

A dict with every setting the package uses.
Source code in src/rag_titan/config.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def get_settings(overrides=None):
    """
    Build the settings dictionary.

    Args:
        overrides: optional dict of settings that win over the
            environment, for example {"chunk_size": 200}.

    Returns:
        A dict with every setting the package uses.
    """
    # read the .env file only when settings are requested, not on import
    if load_dotenv is not None:
        load_dotenv()

    settings = {
        # LLM used by RAG
        "llm_provider": os.getenv("LLM_PROVIDER", "ollama"),
        "llm_model": os.getenv("LLM_MODEL", "gemma3:4b"),
        "ollama_host": os.getenv("OLLAMA_HOST", "http://localhost:11434"),

        # LLM used by agents (needs tool-calling support)
        "agent_model": os.getenv("AGENT_MODEL", "qwen3:4b"),
        "agent_think": _get_bool("AGENT_THINK", True),
        "agent_max_steps": _get_int("AGENT_MAX_STEPS", 5),

        # Embeddings
        "embed_provider": os.getenv("EMBED_PROVIDER", "ollama"),
        "embed_model": os.getenv("EMBED_MODEL", "nomic-embed-text"),

        # Chunking
        "chunker": os.getenv("CHUNKER", "word"),
        "chunk_size": _get_int("CHUNK_SIZE", 150),
        "chunk_overlap": _get_int("CHUNK_OVERLAP", 30),
        "chunk_encoding": os.getenv("CHUNK_ENCODING", "cl100k_base"),  # token chunker only

        # Storage, retrieval and generation
        "vector_store": os.getenv("VECTOR_STORE", "memory"),
        # drop a search result that overlaps a better one by this share ("off" keeps all)
        "dedupe_overlap": _get_overlap("DEDUPE_OVERLAP", 0.5),
        "reranker": os.getenv("RERANKER", "keyword"),
        "generator": os.getenv("GENERATOR", "grounded"),
    }

    if overrides:
        settings.update(overrides)

    return settings