Building blocks

Chat models (LLMs). Add new providers here.

BaseLLM

Common interface for chat models.

Parameters:

Name	Type	Description	Default
model		the model name, for example "gemma3:4b".	required

Source code in src/rag_titan/llm/base_llm.py

class BaseLLM:
    """
    Common interface for chat models.

    Args:
        model: the model name, for example "gemma3:4b".
    """

    def __init__(self, model):
        self.model = model

    def chat(self, messages, tools=None):
        """
        Send a list of messages to the model.

        Args:
            messages: list of {"role": ..., "content": ...} dicts.
            tools: optional list of plain Python functions the model may call.

        Returns:
            A dict:
                {
                    "content": "text reply",
                    "tool_calls": [{"name": "add", "arguments": {"a": 1, "b": 2}}],
                    "message": the assistant message to append to the history,
                }
        """
        raise NotImplementedError

    def tool_result_message(self, tool_call, result):
        """Build the message that sends a tool's result back to the model."""
        raise NotImplementedError

    def ask(self, instructions, user_input):
        """
        Shortcut for a single question with a system instruction.

        Args:
            instructions: the system prompt.
            user_input: the user message.

        Returns:
            The reply text with spaces removed from both ends.
        """
        messages = [
            {"role": "system", "content": instructions},
            {"role": "user", "content": user_input},
        ]
        return self.chat(messages)["content"].strip()

ask(instructions, user_input)

Shortcut for a single question with a system instruction.

Parameters:

Name	Type	Description	Default
instructions		the system prompt.	required
user_input		the user message.	required

Returns:

Type	Description
	The reply text with spaces removed from both ends.

Source code in src/rag_titan/llm/base_llm.py

def ask(self, instructions, user_input):
    """
    Shortcut for a single question with a system instruction.

    Args:
        instructions: the system prompt.
        user_input: the user message.

    Returns:
        The reply text with spaces removed from both ends.
    """
    messages = [
        {"role": "system", "content": instructions},
        {"role": "user", "content": user_input},
    ]
    return self.chat(messages)["content"].strip()

chat(messages, tools=None)

Send a list of messages to the model.

Parameters:

Name	Type	Description	Default
messages		list of {"role": ..., "content": ...} dicts.	required
tools		optional list of plain Python functions the model may call.	None

Returns:

Type	Description
	A dict: { "content": "text reply", "tool_calls": [{"name": "add", "arguments": {"a": 1, "b": 2}}], "message": the assistant message to append to the history, }

Source code in src/rag_titan/llm/base_llm.py

def chat(self, messages, tools=None):
    """
    Send a list of messages to the model.

    Args:
        messages: list of {"role": ..., "content": ...} dicts.
        tools: optional list of plain Python functions the model may call.

    Returns:
        A dict:
            {
                "content": "text reply",
                "tool_calls": [{"name": "add", "arguments": {"a": 1, "b": 2}}],
                "message": the assistant message to append to the history,
            }
    """
    raise NotImplementedError

tool_result_message(tool_call, result)

Build the message that sends a tool's result back to the model.

Source code in src/rag_titan/llm/base_llm.py

def tool_result_message(self, tool_call, result):
    """Build the message that sends a tool's result back to the model."""
    raise NotImplementedError

FunctionLLM

Bases: BaseLLM

A chat model made from a function.

Parameters:

Name	Type	Description	Default
reply		function that takes the messages (a list of {"role", "content"} dicts) and returns the reply text.	required
model		name to show for this model.	'function'

Source code in src/rag_titan/llm/function_llm.py

class FunctionLLM(BaseLLM):
    """
    A chat model made from a function.

    Args:
        reply: function that takes the messages (a list of {"role", "content"}
            dicts) and returns the reply text.
        model: name to show for this model.
    """

    def __init__(self, reply, model="function"):
        if not callable(reply):
            raise ValueError("reply must be a function: messages -> reply text")

        super().__init__(model)
        self.reply = reply

    def chat(self, messages, tools=None):
        """Call the function. Tools are not supported."""
        if tools:
            raise ValueError("FunctionLLM does not support tool calls")

        plain = [{"role": message["role"], "content": message["content"]} for message in messages]
        text = self.reply(plain)
        if not isinstance(text, str):
            raise ValueError(f"The reply function must return text, got {type(text).__name__}")

        message = {"role": "assistant", "content": text}
        return {"content": text, "tool_calls": [], "message": message}

    def tool_result_message(self, tool_call, result):
        """Build a simple tool message (tool calls are not supported)."""
        return {"role": "tool", "content": str(result)}

chat(messages, tools=None)

Call the function. Tools are not supported.

Source code in src/rag_titan/llm/function_llm.py

def chat(self, messages, tools=None):
    """Call the function. Tools are not supported."""
    if tools:
        raise ValueError("FunctionLLM does not support tool calls")

    plain = [{"role": message["role"], "content": message["content"]} for message in messages]
    text = self.reply(plain)
    if not isinstance(text, str):
        raise ValueError(f"The reply function must return text, got {type(text).__name__}")

    message = {"role": "assistant", "content": text}
    return {"content": text, "tool_calls": [], "message": message}

tool_result_message(tool_call, result)

Build a simple tool message (tool calls are not supported).

Source code in src/rag_titan/llm/function_llm.py

def tool_result_message(self, tool_call, result):
    """Build a simple tool message (tool calls are not supported)."""
    return {"role": "tool", "content": str(result)}

OllamaLLM

Bases: BaseLLM

Talks to an Ollama server.

Parameters:

Name	Type	Description	Default
model		Ollama model name, for example "gemma3:4b".	required
host		Ollama server address. None means OLLAMA_HOST, then http://localhost:11434.	None
think		turn on "thinking" for models that support it.	False

Source code in src/rag_titan/llm/ollama_llm.py

class OllamaLLM(BaseLLM):
    """
    Talks to an Ollama server.

    Args:
        model: Ollama model name, for example "gemma3:4b".
        host: Ollama server address. None means OLLAMA_HOST, then
            http://localhost:11434.
        think: turn on "thinking" for models that support it.
    """

    def __init__(self, model, host=None, think=False):
        super().__init__(model)
        self.think = think
        self.host = parse_host(host)

    def chat(self, messages, tools=None):
        """Send messages to Ollama. See BaseLLM.chat() for the reply format."""
        logger.debug("Calling Ollama model '%s'", self.model)

        payload = {
            "model": self.model,
            "stream": False,
            "messages": [clean_message(message) for message in messages],
            "tools": tools_to_schemas(tools),
        }
        # only send "think" when it is on, some models do not support it
        if self.think:
            payload["think"] = True

        try:
            reply = post_json(self.host, "/api/chat", payload)
        except RuntimeError as error:
            raise RuntimeError(
                f"Ollama chat failed for model '{self.model}': {error}"
            ) from error

        message = self._reply_message(reply)
        return {
            "content": message["content"] or "",
            "tool_calls": self._read_tool_calls(message),
            "message": message,
        }

    def tool_result_message(self, tool_call, result):
        """Build the "tool" message Ollama expects after a tool call."""
        return {
            "role": "tool",
            "tool_name": tool_call["name"],
            "content": str(result),
        }

    def _reply_message(self, reply):
        """
        Read the assistant message from Ollama's reply.

        Every known field is present (None when missing), so code can read
        message["content"] safely, like it could with the old client.
        """
        raw = reply.get("message") if isinstance(reply, dict) else None
        if not isinstance(raw, dict):
            raise RuntimeError(
                f"Ollama reply for model '{self.model}' has no message: {reply!r:.200}"
            )

        message = {key: None for key in MESSAGE_FIELDS}
        message.update(clean_message(raw))
        return message

    def _read_tool_calls(self, message):
        """Turn Ollama's tool calls into simple dicts."""
        tool_calls = []

        for call in message["tool_calls"] or []:
            tool_calls.append({
                "name": call["function"]["name"],
                "arguments": dict(call["function"]["arguments"] or {}),
            })

        return tool_calls

chat(messages, tools=None)

Send messages to Ollama. See BaseLLM.chat() for the reply format.

Source code in src/rag_titan/llm/ollama_llm.py

def chat(self, messages, tools=None):
    """Send messages to Ollama. See BaseLLM.chat() for the reply format."""
    logger.debug("Calling Ollama model '%s'", self.model)

    payload = {
        "model": self.model,
        "stream": False,
        "messages": [clean_message(message) for message in messages],
        "tools": tools_to_schemas(tools),
    }
    # only send "think" when it is on, some models do not support it
    if self.think:
        payload["think"] = True

    try:
        reply = post_json(self.host, "/api/chat", payload)
    except RuntimeError as error:
        raise RuntimeError(
            f"Ollama chat failed for model '{self.model}': {error}"
        ) from error

    message = self._reply_message(reply)
    return {
        "content": message["content"] or "",
        "tool_calls": self._read_tool_calls(message),
        "message": message,
    }

tool_result_message(tool_call, result)

Build the "tool" message Ollama expects after a tool call.

Source code in src/rag_titan/llm/ollama_llm.py

def tool_result_message(self, tool_call, result):
    """Build the "tool" message Ollama expects after a tool call."""
    return {
        "role": "tool",
        "tool_name": tool_call["name"],
        "content": str(result),
    }

RecordingLLM

Bases: BaseLLM

Passes every new prompt to a real LLM and remembers the reply.

A prompt asked again gets the remembered reply, like ReplayLLM would give. Models do not always answer the same, so this keeps the recording exactly what a replay will do.

Parameters:

Name	Type	Description	Default
llm		any BaseLLM, for example an OllamaLLM.	required

Source code in src/rag_titan/llm/replay_llm.py

class RecordingLLM(BaseLLM):
    """
    Passes every new prompt to a real LLM and remembers the reply.

    A prompt asked again gets the remembered reply, like ReplayLLM would give.
    Models do not always answer the same, so this keeps the recording exactly
    what a replay will do.

    Args:
        llm: any BaseLLM, for example an OllamaLLM.
    """

    def __init__(self, llm):
        super().__init__(llm.model)
        self.llm = llm
        self.replies = {}

    def chat(self, messages, tools=None):
        """Ask the real LLM (only the first time) and save its reply under the prompt's key."""
        if tools:
            raise ValueError("RecordingLLM records plain replies only, not tool calls")

        key = prompt_key(messages)
        if key not in self.replies:
            self.replies[key] = self.llm.chat(messages)["content"]

        return {"content": self.replies[key], "tool_calls": [], "message": {"role": "assistant"}}

    def tool_result_message(self, tool_call, result):
        """Not used: tool calls are not recorded."""
        return self.llm.tool_result_message(tool_call, result)

chat(messages, tools=None)

Ask the real LLM (only the first time) and save its reply under the prompt's key.

Source code in src/rag_titan/llm/replay_llm.py

def chat(self, messages, tools=None):
    """Ask the real LLM (only the first time) and save its reply under the prompt's key."""
    if tools:
        raise ValueError("RecordingLLM records plain replies only, not tool calls")

    key = prompt_key(messages)
    if key not in self.replies:
        self.replies[key] = self.llm.chat(messages)["content"]

    return {"content": self.replies[key], "tool_calls": [], "message": {"role": "assistant"}}

tool_result_message(tool_call, result)

Not used: tool calls are not recorded.

Source code in src/rag_titan/llm/replay_llm.py

def tool_result_message(self, tool_call, result):
    """Not used: tool calls are not recorded."""
    return self.llm.tool_result_message(tool_call, result)

ReplayLLM

Bases: BaseLLM

Answers from recorded replies.

Parameters:

Name	Type	Description	Default
replies		dict of {prompt key: reply text}, e.g. RecordingLLM.replies.	required
model		name to show, e.g. the model the replies came from.	'replay'

Source code in src/rag_titan/llm/replay_llm.py

class ReplayLLM(BaseLLM):
    """
    Answers from recorded replies.

    Args:
        replies: dict of {prompt key: reply text}, e.g. RecordingLLM.replies.
        model: name to show, e.g. the model the replies came from.
    """

    def __init__(self, replies, model="replay"):
        if not isinstance(replies, dict):
            raise ValueError("replies must be a dict of {prompt key: reply text}")

        super().__init__(model)
        self.replies = replies

    def chat(self, messages, tools=None):
        """Return the recorded reply for exactly these messages."""
        key = prompt_key(messages)
        if key not in self.replies:
            raise ValueError(
                f"No recorded reply for this prompt (key {key}). The question, the "
                "documents or the settings differ from the recording; record it again."
            )

        return {"content": self.replies[key], "tool_calls": [], "message": {"role": "assistant"}}

    def tool_result_message(self, tool_call, result):
        """Build a simple tool message (tool calls are not replayed)."""
        return {"role": "tool", "content": str(result)}

chat(messages, tools=None)

Return the recorded reply for exactly these messages.

Source code in src/rag_titan/llm/replay_llm.py

def chat(self, messages, tools=None):
    """Return the recorded reply for exactly these messages."""
    key = prompt_key(messages)
    if key not in self.replies:
        raise ValueError(
            f"No recorded reply for this prompt (key {key}). The question, the "
            "documents or the settings differ from the recording; record it again."
        )

    return {"content": self.replies[key], "tool_calls": [], "message": {"role": "assistant"}}

tool_result_message(tool_call, result)

Build a simple tool message (tool calls are not replayed).

Source code in src/rag_titan/llm/replay_llm.py

def tool_result_message(self, tool_call, result):
    """Build a simple tool message (tool calls are not replayed)."""
    return {"role": "tool", "content": str(result)}

File loaders that turn files into plain text documents.

BaseLoader

Reads one kind of file and turns it into a document dict.

Source code in src/rag_titan/loaders/base_loader.py

class BaseLoader:
    """Reads one kind of file and turns it into a document dict."""

    # file extensions this loader can read, for example [".pdf"]
    extensions = []

    def load(self, path):
        """
        Read a file and return a document dict.

        Args:
            path: path to the file.

        Returns:
            {"source": file name, "file_type": ".pdf", "text": all text}
        """
        path = str(path)

        if not os.path.isfile(path):
            raise FileNotFoundError(f"File not found: {path}")

        return {
            "source": os.path.basename(path),
            "file_type": os.path.splitext(path)[1].lower(),
            "text": self.extract_text(path),
        }

    def extract_text(self, path):
        """Return all text in the file as one string."""
        raise NotImplementedError

extract_text(path)

Return all text in the file as one string.

Source code in src/rag_titan/loaders/base_loader.py

def extract_text(self, path):
    """Return all text in the file as one string."""
    raise NotImplementedError

load(path)

Read a file and return a document dict.

Parameters:

Name	Type	Description	Default
path		path to the file.	required

Returns:

Type	Description
	{"source": file name, "file_type": ".pdf", "text": all text}

Source code in src/rag_titan/loaders/base_loader.py

def load(self, path):
    """
    Read a file and return a document dict.

    Args:
        path: path to the file.

    Returns:
        {"source": file name, "file_type": ".pdf", "text": all text}
    """
    path = str(path)

    if not os.path.isfile(path):
        raise FileNotFoundError(f"File not found: {path}")

    return {
        "source": os.path.basename(path),
        "file_type": os.path.splitext(path)[1].lower(),
        "text": self.extract_text(path),
    }

DocxLoader

Bases: BaseLoader

Reads the text of every paragraph in a .docx file.

Paragraphs are separated by a blank line, and Word headings become Markdown headings ("Heading 2" -> "## "), so chunkers can see the document's structure.

Source code in src/rag_titan/loaders/docx_loader.py

class DocxLoader(BaseLoader):
    """
    Reads the text of every paragraph in a .docx file.

    Paragraphs are separated by a blank line, and Word headings become
    Markdown headings ("Heading 2" -> "## "), so chunkers can see the
    document's structure.
    """

    extensions = [".docx"]

    def extract_text(self, path):
        """Return all non-empty paragraphs, separated by blank lines."""
        # imported here, so the package also loads where python-docx is missing (the browser)
        from docx import Document

        blocks = []
        for para in Document(path).paragraphs:
            text = para.text.strip()
            if not text:
                continue

            level = heading_level(para.style.name if para.style is not None else "")
            blocks.append("#" * level + " " + text if level else text)

        return "\n\n".join(blocks)

extract_text(path)

Return all non-empty paragraphs, separated by blank lines.

Source code in src/rag_titan/loaders/docx_loader.py

def extract_text(self, path):
    """Return all non-empty paragraphs, separated by blank lines."""
    # imported here, so the package also loads where python-docx is missing (the browser)
    from docx import Document

    blocks = []
    for para in Document(path).paragraphs:
        text = para.text.strip()
        if not text:
            continue

        level = heading_level(para.style.name if para.style is not None else "")
        blocks.append("#" * level + " " + text if level else text)

    return "\n\n".join(blocks)

PdfLoader

Bases: BaseLoader

Reads the text of every page in a .pdf file.

Source code in src/rag_titan/loaders/pdf_loader.py

class PdfLoader(BaseLoader):
    """Reads the text of every page in a .pdf file."""

    extensions = [".pdf"]

    def extract_text(self, path):
        """Return the text of all pages, separated by a page break ("\\f")."""
        # imported here, so the package also loads where pypdf is missing (the browser)
        from pypdf import PdfReader

        reader = PdfReader(path)

        # some pages (for example scanned images) have no text, so use ""
        pages = [page.extract_text() or "" for page in reader.pages]
        return PAGE_BREAK.join(pages)

extract_text(path)

Return the text of all pages, separated by a page break ("\f").

Source code in src/rag_titan/loaders/pdf_loader.py

def extract_text(self, path):
    """Return the text of all pages, separated by a page break ("\\f")."""
    # imported here, so the package also loads where pypdf is missing (the browser)
    from pypdf import PdfReader

    reader = PdfReader(path)

    # some pages (for example scanned images) have no text, so use ""
    pages = [page.extract_text() or "" for page in reader.pages]
    return PAGE_BREAK.join(pages)

TextLoader

Bases: BaseLoader

Reads .txt and .md files as they are, so Markdown headings are kept.

Source code in src/rag_titan/loaders/text_loader.py

class TextLoader(BaseLoader):
    """Reads .txt and .md files as they are, so Markdown headings are kept."""

    extensions = [".txt", ".md", ".markdown"]

    def extract_text(self, path):
        """Return the file's text. Bytes that are not valid UTF-8 become "�" instead of failing."""
        with open(path, encoding="utf-8", errors="replace") as file:
            return file.read()

extract_text(path)

Return the file's text. Bytes that are not valid UTF-8 become "�" instead of failing.

Source code in src/rag_titan/loaders/text_loader.py

def extract_text(self, path):
    """Return the file's text. Bytes that are not valid UTF-8 become "�" instead of failing."""
    with open(path, encoding="utf-8", errors="replace") as file:
        return file.read()

get_loader(path)

Find the loader that can read this file.

Raises:

Type	Description
ValueError	when no loader supports the file extension.

Source code in src/rag_titan/loaders/loader_registry.py

def get_loader(path):
    """
    Find the loader that can read this file.

    Raises:
        ValueError: when no loader supports the file extension.
    """
    extension = os.path.splitext(str(path))[1].lower()

    for loader in LOADERS:
        if extension in loader.extensions:
            return loader

    supported = [ext for loader in LOADERS for ext in loader.extensions]
    raise ValueError(f"Unsupported file type '{extension}'. Supported: {supported}")

load_document(path)

Load any supported file and return its document dict.

Source code in src/rag_titan/loaders/loader_registry.py

def load_document(path):
    """Load any supported file and return its document dict."""
    return get_loader(path).load(path)

Chunkers that split documents into small pieces.

Each family of strategies lives in its own sub-folder

fixed_size/ characters, words, tokens, recursive separators (handbook section 2) structure_aware/ sentences, paragraphs, headings, pages (handbook section 3)

Everything is also exported here, so from rag_titan.chunking import WordChunker works.

BaseChunker

Splits documents into smaller pieces called chunks.

Source code in src/rag_titan/chunking/base_chunker.py

class BaseChunker:
    """Splits documents into smaller pieces called chunks."""

    def chunk(self, documents):
        """
        Turn a list of document dicts into a list of chunk dicts.

        Args:
            documents: list of {"source": ..., "text": ...} dicts.

        Returns:
            List of {"id": "chunk_0", "source": ..., "text": ...,
            "start_offset": 0, "end_offset": 120} dicts.
            `document["text"][start_offset:end_offset]` is where the chunk comes
            from (the offsets are None if it cannot be found).
            When known, chunks also get "page_numbers" (PDFs) and
            "section_path" (documents with headings).
        """
        chunks = []

        for document in documents:
            locations = Locations(document["text"])
            for text, start, end in self.split_spans(document["text"]):
                chunks.append({
                    "id": f"chunk_{len(chunks)}",
                    "source": document["source"],
                    "text": text,
                    "start_offset": start,
                    "end_offset": end,
                    **locations.describe(start, end),
                })

        return chunks

    def split_spans(self, text):
        """
        Return (chunk_text, start, end) for every chunk of `text`.

        This default searches for each chunk from split_text() in the text.
        Chunkers that know their exact positions override it.
        """
        return located_spans(text, self.split_text(text))

    def split_text(self, text):
        """Return a list of text pieces."""
        raise NotImplementedError

chunk(documents)

Turn a list of document dicts into a list of chunk dicts.

Parameters:

Name	Type	Description	Default
documents		list of {"source": ..., "text": ...} dicts.	required

Returns:

Type	Description
	List of {"id": "chunk_0", "source": ..., "text": ...,
	"start_offset": 0, "end_offset": 120} dicts.
	document["text"][start_offset:end_offset] is where the chunk comes
	from (the offsets are None if it cannot be found).
	When known, chunks also get "page_numbers" (PDFs) and
	"section_path" (documents with headings).

Source code in src/rag_titan/chunking/base_chunker.py

def chunk(self, documents):
    """
    Turn a list of document dicts into a list of chunk dicts.

    Args:
        documents: list of {"source": ..., "text": ...} dicts.

    Returns:
        List of {"id": "chunk_0", "source": ..., "text": ...,
        "start_offset": 0, "end_offset": 120} dicts.
        `document["text"][start_offset:end_offset]` is where the chunk comes
        from (the offsets are None if it cannot be found).
        When known, chunks also get "page_numbers" (PDFs) and
        "section_path" (documents with headings).
    """
    chunks = []

    for document in documents:
        locations = Locations(document["text"])
        for text, start, end in self.split_spans(document["text"]):
            chunks.append({
                "id": f"chunk_{len(chunks)}",
                "source": document["source"],
                "text": text,
                "start_offset": start,
                "end_offset": end,
                **locations.describe(start, end),
            })

    return chunks

split_spans(text)

Return (chunk_text, start, end) for every chunk of text.

This default searches for each chunk from split_text() in the text. Chunkers that know their exact positions override it.

Source code in src/rag_titan/chunking/base_chunker.py

def split_spans(self, text):
    """
    Return (chunk_text, start, end) for every chunk of `text`.

    This default searches for each chunk from split_text() in the text.
    Chunkers that know their exact positions override it.
    """
    return located_spans(text, self.split_text(text))

split_text(text)

Return a list of text pieces.

Source code in src/rag_titan/chunking/base_chunker.py

def split_text(self, text):
    """Return a list of text pieces."""
    raise NotImplementedError

CharacterChunker

Bases: BaseChunker

Splits text into fixed-size character windows that overlap. Example: size 1000, overlap 200 -> each new chunk starts 800 characters later.

The text is kept exactly as it is (spaces and newlines included), so a chunk may start or end in the middle of a word.

Parameters:

Name	Type	Description	Default
chunk_size		number of characters in each chunk.	1000
overlap		number of characters shared by two neighbouring chunks.	200

Source code in src/rag_titan/chunking/fixed_size/character_chunker.py

class CharacterChunker(BaseChunker):
    """
    Splits text into fixed-size character windows that overlap.
    Example: size 1000, overlap 200 -> each new chunk starts 800 characters later.

    The text is kept exactly as it is (spaces and newlines included), so a
    chunk may start or end in the middle of a word.

    Args:
        chunk_size: number of characters in each chunk.
        overlap: number of characters shared by two neighbouring chunks.
    """

    def __init__(self, chunk_size=1000, overlap=200):
        check_window_size(chunk_size, overlap)
        self.chunk_size = chunk_size
        self.overlap = overlap

    def split_text(self, text):
        """Return overlapping pieces of `chunk_size` characters each."""
        return [chunk for chunk, _, _ in self.split_spans(text)]

    def split_spans(self, text):
        """(chunk, start, end) for each window; window k starts at k * (chunk_size - overlap)."""
        # blank text has nothing worth indexing, so it gives no chunks
        if not text.strip():
            return []

        windows = fixed_windows(range(len(text)), self.chunk_size, self.overlap)
        return [(text[w[0]:w[-1] + 1], w[0], w[-1] + 1) for w in windows]

split_spans(text)

(chunk, start, end) for each window; window k starts at k * (chunk_size - overlap).

Source code in src/rag_titan/chunking/fixed_size/character_chunker.py

def split_spans(self, text):
    """(chunk, start, end) for each window; window k starts at k * (chunk_size - overlap)."""
    # blank text has nothing worth indexing, so it gives no chunks
    if not text.strip():
        return []

    windows = fixed_windows(range(len(text)), self.chunk_size, self.overlap)
    return [(text[w[0]:w[-1] + 1], w[0], w[-1] + 1) for w in windows]

split_text(text)

Return overlapping pieces of chunk_size characters each.

Source code in src/rag_titan/chunking/fixed_size/character_chunker.py

def split_text(self, text):
    """Return overlapping pieces of `chunk_size` characters each."""
    return [chunk for chunk, _, _ in self.split_spans(text)]

HeadingChunker

Bases: PieceChunker

Splits Markdown text at its headings, so a chunk never mixes two sections. Inside a section, whole paragraphs are packed up to chunk_size (then sentences, then words, when one is too big).

Parameters:

Name	Type	Description	Default
chunk_size		largest chunk, in characters (or tokens with a tokenizer), including the heading path.	1000
overlap		whole paragraphs repeated inside a section, up to this size.	0
include_path		start each chunk with its heading path, for example "Leave policy > Sick leave". The path is skipped when it would take more than half of chunk_size.	True
tokenizer		optional, counts tokens instead of characters.	None

Source code in src/rag_titan/chunking/structure_aware/heading_chunker.py

class HeadingChunker(PieceChunker):
    """
    Splits Markdown text at its headings, so a chunk never mixes two sections.
    Inside a section, whole paragraphs are packed up to `chunk_size` (then
    sentences, then words, when one is too big).

    Args:
        chunk_size: largest chunk, in characters (or tokens with a tokenizer),
            including the heading path.
        overlap: whole paragraphs repeated inside a section, up to this size.
        include_path: start each chunk with its heading path, for example
            "Leave policy > Sick leave". The path is skipped when it would take
            more than half of chunk_size.
        tokenizer: optional, counts tokens instead of characters.
    """

    def __init__(self, chunk_size=1000, overlap=0, include_path=True, tokenizer=None):
        super().__init__(chunk_size, overlap, tokenizer)
        self.include_path = include_path

    def split_text(self, text):
        """Return chunks that each stay inside one section."""
        return [chunk for chunk, _, _ in self.split_spans(text)]

    def split_spans(self, text):
        """(chunk with heading path, start, end); start/end point at the section text."""
        if not text.strip():
            return []

        spans = []
        for path, body, start in section_spans(text, keep_heading_lines=not self.include_path):
            spans.extend(self.section_chunks(path, body, start))
        return spans

    def section_chunks(self, path, body, start):
        """Chunk one section's body; put the heading path in front of each chunk."""
        prefix = " > ".join(path) + "\n\n" if self.include_path and path else ""
        room = self.chunk_size - self.length(prefix)

        if room < self.chunk_size / 2:
            prefix, room = "", self.chunk_size   # a very long path would crowd out the text

        inner = ParagraphChunker(room, min(self.overlap, room - 1), self.tokenizer)
        return [(prefix + chunk, s, e) for chunk, s, e in shift(inner.split_spans(body), start)]

section_chunks(path, body, start)

Chunk one section's body; put the heading path in front of each chunk.

Source code in src/rag_titan/chunking/structure_aware/heading_chunker.py

def section_chunks(self, path, body, start):
    """Chunk one section's body; put the heading path in front of each chunk."""
    prefix = " > ".join(path) + "\n\n" if self.include_path and path else ""
    room = self.chunk_size - self.length(prefix)

    if room < self.chunk_size / 2:
        prefix, room = "", self.chunk_size   # a very long path would crowd out the text

    inner = ParagraphChunker(room, min(self.overlap, room - 1), self.tokenizer)
    return [(prefix + chunk, s, e) for chunk, s, e in shift(inner.split_spans(body), start)]

split_spans(text)

(chunk with heading path, start, end); start/end point at the section text.

Source code in src/rag_titan/chunking/structure_aware/heading_chunker.py

def split_spans(self, text):
    """(chunk with heading path, start, end); start/end point at the section text."""
    if not text.strip():
        return []

    spans = []
    for path, body, start in section_spans(text, keep_heading_lines=not self.include_path):
        spans.extend(self.section_chunks(path, body, start))
    return spans

split_text(text)

Return chunks that each stay inside one section.

Source code in src/rag_titan/chunking/structure_aware/heading_chunker.py

def split_text(self, text):
    """Return chunks that each stay inside one section."""
    return [chunk for chunk, _, _ in self.split_spans(text)]

PageChunker

Bases: PieceChunker

Chunks each page on its own, so a chunk never spans two pages; good for citing page numbers. A page that fits is one chunk; a bigger page is cut into whole paragraphs (then sentences, then words). Empty pages are skipped. Text without page breaks (Word, Markdown) is treated as one page.

Parameters:

Name	Type	Description	Default
chunk_size		largest chunk, in characters (or tokens with a tokenizer).	2000
overlap		whole paragraphs repeated inside a page, up to this size.	0
tokenizer		optional, counts tokens instead of characters.	None

Source code in src/rag_titan/chunking/structure_aware/page_chunker.py

class PageChunker(PieceChunker):
    """
    Chunks each page on its own, so a chunk never spans two pages; good for
    citing page numbers. A page that fits is one chunk; a bigger page is cut
    into whole paragraphs (then sentences, then words). Empty pages are skipped.
    Text without page breaks (Word, Markdown) is treated as one page.

    Args:
        chunk_size: largest chunk, in characters (or tokens with a tokenizer).
        overlap: whole paragraphs repeated inside a page, up to this size.
        tokenizer: optional, counts tokens instead of characters.
    """

    def __init__(self, chunk_size=2000, overlap=0, tokenizer=None):
        super().__init__(chunk_size, overlap, tokenizer)
        self.paragraphs = ParagraphChunker(chunk_size, overlap, tokenizer)

    def split_text(self, text):
        """Return chunks that each stay inside one page."""
        return [chunk for chunk, _, _ in self.split_spans(text)]

    def split_spans(self, text):
        """(chunk, start, end) for every chunk; positions count from the start of `text`."""
        spans, page_start = [], 0
        for page in text.split(PAGE_BREAK):
            spans.extend(shift(self.paragraphs.split_spans(page), page_start))
            page_start += len(page) + len(PAGE_BREAK)
        return spans

split_spans(text)

(chunk, start, end) for every chunk; positions count from the start of text.

Source code in src/rag_titan/chunking/structure_aware/page_chunker.py

def split_spans(self, text):
    """(chunk, start, end) for every chunk; positions count from the start of `text`."""
    spans, page_start = [], 0
    for page in text.split(PAGE_BREAK):
        spans.extend(shift(self.paragraphs.split_spans(page), page_start))
        page_start += len(page) + len(PAGE_BREAK)
    return spans

split_text(text)

Return chunks that each stay inside one page.

Source code in src/rag_titan/chunking/structure_aware/page_chunker.py

def split_text(self, text):
    """Return chunks that each stay inside one page."""
    return [chunk for chunk, _, _ in self.split_spans(text)]

ParagraphChunker

Bases: PieceChunker

Packs whole paragraphs (separated by a blank line) into chunks of at most chunk_size. A paragraph that is too long is cut into sentences, and a sentence that is still too long is cut at words.

Parameters:

Name	Type	Description	Default
chunk_size		largest chunk, in characters (or tokens with a tokenizer).	1000
overlap		whole paragraphs repeated from the previous chunk, up to this size.	0
tokenizer		optional, counts tokens instead of characters.	None

Source code in src/rag_titan/chunking/structure_aware/paragraph_chunker.py

class ParagraphChunker(PieceChunker):
    """
    Packs whole paragraphs (separated by a blank line) into chunks of at most
    `chunk_size`. A paragraph that is too long is cut into sentences, and a
    sentence that is still too long is cut at words.

    Args:
        chunk_size: largest chunk, in characters (or tokens with a tokenizer).
        overlap: whole paragraphs repeated from the previous chunk, up to this size.
        tokenizer: optional, counts tokens instead of characters.
    """

    def __init__(self, chunk_size=1000, overlap=0, tokenizer=None):
        super().__init__(chunk_size, overlap, tokenizer)

    def pieces(self, text):
        """Paragraphs, with too-long paragraphs cut into sentences (then words)."""
        return self.fit(split_paragraphs(text), self.sentence_pieces)

    def sentence_pieces(self, paragraph):
        """Sentences of one paragraph, with too-long sentences cut at words."""
        return self.fit(split_sentences(paragraph), self.word_splitter.split_pieces)

pieces(text)

Paragraphs, with too-long paragraphs cut into sentences (then words).

Source code in src/rag_titan/chunking/structure_aware/paragraph_chunker.py

def pieces(self, text):
    """Paragraphs, with too-long paragraphs cut into sentences (then words)."""
    return self.fit(split_paragraphs(text), self.sentence_pieces)

sentence_pieces(paragraph)

Sentences of one paragraph, with too-long sentences cut at words.

Source code in src/rag_titan/chunking/structure_aware/paragraph_chunker.py

def sentence_pieces(self, paragraph):
    """Sentences of one paragraph, with too-long sentences cut at words."""
    return self.fit(split_sentences(paragraph), self.word_splitter.split_pieces)

RecursiveChunker

Bases: BaseChunker

Splits text on paragraphs, then lines, sentences and words, packing the pieces into chunks of at most chunk_size characters (or tokens). Example: size 1000, overlap 200 -> paragraphs stay whole when they fit.

Parameters:

Name	Type	Description	Default
chunk_size		largest chunk, in characters (or tokens with a tokenizer).	1000
overlap		how much of the end of a chunk is repeated at the start of the next one. Only whole pieces are repeated, so it may be less.	200
separators		separators to try, biggest first. "" means a hard cut.	SEPARATORS
tokenizer		optional, any object with encode(text) -> ids and decode(ids) -> text. When given, sizes count tokens.	None

Source code in src/rag_titan/chunking/fixed_size/recursive_chunker.py

class RecursiveChunker(BaseChunker):
    """
    Splits text on paragraphs, then lines, sentences and words, packing the
    pieces into chunks of at most `chunk_size` characters (or tokens).
    Example: size 1000, overlap 200 -> paragraphs stay whole when they fit.

    Args:
        chunk_size: largest chunk, in characters (or tokens with a tokenizer).
        overlap: how much of the end of a chunk is repeated at the start of
            the next one. Only whole pieces are repeated, so it may be less.
        separators: separators to try, biggest first. "" means a hard cut.
        tokenizer: optional, any object with encode(text) -> ids and
            decode(ids) -> text. When given, sizes count tokens.
    """

    def __init__(self, chunk_size=1000, overlap=200, separators=SEPARATORS, tokenizer=None):
        check_window_size(chunk_size, overlap)

        if not separators:
            raise ValueError("separators must contain at least one separator")

        has_methods = hasattr(tokenizer, "encode") and hasattr(tokenizer, "decode")
        if tokenizer is not None and not has_methods:
            raise ValueError("tokenizer must have encode(text) and decode(ids) methods")

        self.chunk_size = chunk_size
        self.overlap = overlap
        self.separators = tuple(separators)
        self.tokenizer = tokenizer

    def length(self, text):
        """Size of `text` in characters, or in tokens when a tokenizer is set."""
        if self.tokenizer is None:
            return len(text)
        return len(self.tokenizer.encode(text))

    def split_text(self, text):
        """Return chunks of at most `chunk_size`, split on the biggest separators possible."""
        return [chunk for chunk, _, _ in self.split_spans(text)]

    def split_spans(self, text):
        """(chunk, start, end) for every chunk; exact, because the pieces join back into text."""
        # blank text has nothing worth indexing, so it gives no chunks
        if not text.strip():
            return []

        pieces = self._split(text, self.separators)
        groups = pack([self.length(piece) for piece in pieces], self.chunk_size, self.overlap)
        spans = group_spans(text, pieces, groups)
        if spans is None:   # a tokenizer changed the text, so search for the chunks instead
            chunks = [chunk.strip() for chunk in join_groups(pieces, groups) if chunk.strip()]
            return located_spans(text, chunks)
        return spans

    def split_pieces(self, text):
        """
        Cut `text` into pieces that each fit in chunk_size, without packing them.
        Joining the pieces gives back `text` exactly. Other chunkers use this
        as a fallback for pieces that are too big.
        """
        return self._split(text, self.separators)

    def _split(self, text, separators):
        """Break `text` into pieces that each fit in chunk_size."""
        if self.length(text) <= self.chunk_size:
            return [text]

        for index, separator in enumerate(separators):
            if separator == "":
                return self._hard_cut(text)

            # use the biggest separator found; smaller ones handle what is still too big
            if separator in text:
                pieces = []
                for part in split_keep_separator(text, separator):
                    pieces.extend(self._split(part, separators[index + 1:]))
                return pieces

        # no separator matched and no "" fallback was given
        return self._hard_cut(text)

    def _hard_cut(self, text):
        """Last resort: cut `text` into exact chunk_size pieces."""
        if self.tokenizer is None:
            return fixed_windows(text, self.chunk_size)

        ids = list(self.tokenizer.encode(text))
        return [self.tokenizer.decode(window) for window in fixed_windows(ids, self.chunk_size)]

length(text)

Size of text in characters, or in tokens when a tokenizer is set.

Source code in src/rag_titan/chunking/fixed_size/recursive_chunker.py

def length(self, text):
    """Size of `text` in characters, or in tokens when a tokenizer is set."""
    if self.tokenizer is None:
        return len(text)
    return len(self.tokenizer.encode(text))

split_pieces(text)

Cut text into pieces that each fit in chunk_size, without packing them. Joining the pieces gives back text exactly. Other chunkers use this as a fallback for pieces that are too big.

Source code in src/rag_titan/chunking/fixed_size/recursive_chunker.py

def split_pieces(self, text):
    """
    Cut `text` into pieces that each fit in chunk_size, without packing them.
    Joining the pieces gives back `text` exactly. Other chunkers use this
    as a fallback for pieces that are too big.
    """
    return self._split(text, self.separators)

split_spans(text)

(chunk, start, end) for every chunk; exact, because the pieces join back into text.

Source code in src/rag_titan/chunking/fixed_size/recursive_chunker.py

def split_spans(self, text):
    """(chunk, start, end) for every chunk; exact, because the pieces join back into text."""
    # blank text has nothing worth indexing, so it gives no chunks
    if not text.strip():
        return []

    pieces = self._split(text, self.separators)
    groups = pack([self.length(piece) for piece in pieces], self.chunk_size, self.overlap)
    spans = group_spans(text, pieces, groups)
    if spans is None:   # a tokenizer changed the text, so search for the chunks instead
        chunks = [chunk.strip() for chunk in join_groups(pieces, groups) if chunk.strip()]
        return located_spans(text, chunks)
    return spans

split_text(text)

Return chunks of at most chunk_size, split on the biggest separators possible.

Source code in src/rag_titan/chunking/fixed_size/recursive_chunker.py

def split_text(self, text):
    """Return chunks of at most `chunk_size`, split on the biggest separators possible."""
    return [chunk for chunk, _, _ in self.split_spans(text)]

SentenceChunker

Bases: PieceChunker

Packs whole sentences into chunks of at most chunk_size. A sentence is only cut (at words) when it is longer than chunk_size. Example: size 500, overlap 100 -> chunks end at sentence boundaries and repeat the last sentence or two of the previous chunk.

Parameters:

Name	Type	Description	Default
chunk_size		largest chunk, in characters (or tokens with a tokenizer).	500
overlap		whole sentences repeated from the previous chunk, up to this size.	100
tokenizer		optional, counts tokens instead of characters.	None

Source code in src/rag_titan/chunking/structure_aware/sentence_chunker.py

class SentenceChunker(PieceChunker):
    """
    Packs whole sentences into chunks of at most `chunk_size`.
    A sentence is only cut (at words) when it is longer than chunk_size.
    Example: size 500, overlap 100 -> chunks end at sentence boundaries and
    repeat the last sentence or two of the previous chunk.

    Args:
        chunk_size: largest chunk, in characters (or tokens with a tokenizer).
        overlap: whole sentences repeated from the previous chunk, up to this size.
        tokenizer: optional, counts tokens instead of characters.
    """

    def __init__(self, chunk_size=500, overlap=100, tokenizer=None):
        super().__init__(chunk_size, overlap, tokenizer)

    def pieces(self, text):
        """Sentences, with too-long sentences cut at words."""
        return self.fit(split_sentences(text), self.word_splitter.split_pieces)

pieces(text)

Sentences, with too-long sentences cut at words.

Source code in src/rag_titan/chunking/structure_aware/sentence_chunker.py

def pieces(self, text):
    """Sentences, with too-long sentences cut at words."""
    return self.fit(split_sentences(text), self.word_splitter.split_pieces)

TokenChunker

Bases: BaseChunker

Splits text into fixed-size windows of model tokens that overlap. Use the same tokenizer as your embedding model, so no chunk is ever longer than the model accepts. Example: size 256, overlap 32 -> each new chunk starts 224 tokens later.

Parameters:

Name	Type	Description	Default
chunk_size		number of tokens in each chunk.	256
overlap		number of tokens shared by two neighbouring chunks.	32
encoding		tiktoken encoding name, used when no tokenizer is given.	'cl100k_base'
tokenizer		your own tokenizer, any object with encode(text) -> ids and decode(ids) -> text. For a Hugging Face tokenizer, wrap it so encode() calls encode(text, add_special_tokens=False).	None

Source code in src/rag_titan/chunking/fixed_size/token_chunker.py

class TokenChunker(BaseChunker):
    """
    Splits text into fixed-size windows of model tokens that overlap.
    Use the same tokenizer as your embedding model, so no chunk is ever
    longer than the model accepts.
    Example: size 256, overlap 32 -> each new chunk starts 224 tokens later.

    Args:
        chunk_size: number of tokens in each chunk.
        overlap: number of tokens shared by two neighbouring chunks.
        encoding: tiktoken encoding name, used when no tokenizer is given.
        tokenizer: your own tokenizer, any object with encode(text) -> ids
            and decode(ids) -> text. For a Hugging Face tokenizer, wrap it so
            encode() calls encode(text, add_special_tokens=False).
    """

    def __init__(self, chunk_size=256, overlap=32, encoding="cl100k_base", tokenizer=None):
        check_window_size(chunk_size, overlap)

        if tokenizer is None:
            tokenizer = load_tiktoken(encoding)
        elif not (hasattr(tokenizer, "encode") and hasattr(tokenizer, "decode")):
            raise ValueError("tokenizer must have encode(text) and decode(ids) methods")

        self.chunk_size = chunk_size
        self.overlap = overlap
        self.tokenizer = tokenizer

    def split_text(self, text):
        """Return overlapping pieces of `chunk_size` tokens each."""
        # blank text has nothing worth indexing, so it gives no chunks
        if not text.strip():
            return []

        ids = list(self.tokenizer.encode(text))
        windows = fixed_windows(ids, self.chunk_size, self.overlap)
        return [self.tokenizer.decode(window) for window in windows]

split_text(text)

Return overlapping pieces of chunk_size tokens each.

Source code in src/rag_titan/chunking/fixed_size/token_chunker.py

def split_text(self, text):
    """Return overlapping pieces of `chunk_size` tokens each."""
    # blank text has nothing worth indexing, so it gives no chunks
    if not text.strip():
        return []

    ids = list(self.tokenizer.encode(text))
    windows = fixed_windows(ids, self.chunk_size, self.overlap)
    return [self.tokenizer.decode(window) for window in windows]

WordChunker

Bases: BaseChunker

Splits text into fixed-size word windows that overlap. Example: size 150, overlap 30 -> each new chunk starts 120 words later.

Parameters:

Name	Type	Description	Default
chunk_size		number of words in each chunk.	150
overlap		number of words shared by two neighbouring chunks.	30

Source code in src/rag_titan/chunking/fixed_size/word_chunker.py

class WordChunker(BaseChunker):
    """
    Splits text into fixed-size word windows that overlap.
    Example: size 150, overlap 30 -> each new chunk starts 120 words later.

    Args:
        chunk_size: number of words in each chunk.
        overlap: number of words shared by two neighbouring chunks.
    """

    def __init__(self, chunk_size=150, overlap=30):
        if chunk_size <= 0:
            raise ValueError("chunk_size must be greater than 0")

        if overlap < 0 or overlap >= chunk_size:
            raise ValueError("overlap must be between 0 and chunk_size - 1")

        self.chunk_size = chunk_size
        self.overlap = overlap

    def split_text(self, text):
        """Return overlapping pieces of `chunk_size` words each."""
        return [chunk for chunk, _, _ in self.split_spans(text)]

    def split_spans(self, text):
        """(chunk, start, end): from the first word's start to the last word's end."""
        words = list(re.finditer(r"\S+", text))   # the same words as text.split()
        windows = fixed_windows(words, self.chunk_size, self.overlap)
        return [(" ".join(word.group() for word in window), window[0].start(), window[-1].end())
                for window in windows]

split_spans(text)

(chunk, start, end): from the first word's start to the last word's end.

Source code in src/rag_titan/chunking/fixed_size/word_chunker.py

def split_spans(self, text):
    """(chunk, start, end): from the first word's start to the last word's end."""
    words = list(re.finditer(r"\S+", text))   # the same words as text.split()
    windows = fixed_windows(words, self.chunk_size, self.overlap)
    return [(" ".join(word.group() for word in window), window[0].start(), window[-1].end())
            for window in windows]

split_text(text)

Return overlapping pieces of chunk_size words each.

Source code in src/rag_titan/chunking/fixed_size/word_chunker.py

def split_text(self, text):
    """Return overlapping pieces of `chunk_size` words each."""
    return [chunk for chunk, _, _ in self.split_spans(text)]

count_windows(n, size, overlap=0)

Count the windows fixed_windows() makes for n units, without making them.

Overlap repeats text, so it grows the index (handbook 2.4 "index amplification"). Compare count_windows(n, size, overlap) with count_windows(n, size, 0) to see the cost before you index anything. Formula for n > size: 1 + ceil((n - size) / (size - overlap)).

Parameters:

Name	Type	Description	Default
n		number of units (characters, words or tokens) in the text.	required
size		number of units in each window.	required
overlap		number of units shared by two neighbouring windows.	0

Returns:

Type	Description
	The number of windows (0 for empty text).

Source code in src/rag_titan/chunking/fixed_size/fixed_windows.py

def count_windows(n, size, overlap=0):
    """
    Count the windows fixed_windows() makes for `n` units, without making them.

    Overlap repeats text, so it grows the index (handbook 2.4 "index
    amplification"). Compare count_windows(n, size, overlap) with
    count_windows(n, size, 0) to see the cost before you index anything.
    Formula for n > size: 1 + ceil((n - size) / (size - overlap)).

    Args:
        n: number of units (characters, words or tokens) in the text.
        size: number of units in each window.
        overlap: number of units shared by two neighbouring windows.

    Returns:
        The number of windows (0 for empty text).
    """
    check_window_size(size, overlap)

    if not isinstance(n, int) or n < 0:
        raise ValueError("n must be a whole number, 0 or more")

    if n <= size:
        return min(n, 1)

    step = size - overlap
    # whole-number ceiling of (n - size) / step
    return 1 + (n - size + step - 1) // step

Embedders that turn text into vectors.

BaseEmbedder

Turns text into vectors (lists of numbers) that capture meaning.

Parameters:

Name	Type	Description	Default
model		the embedding model name.	required

Source code in src/rag_titan/embeddings/base_embedder.py

class BaseEmbedder:
    """
    Turns text into vectors (lists of numbers) that capture meaning.

    Args:
        model: the embedding model name.
    """

    def __init__(self, model):
        self.model = model

    def embed(self, text):
        """Return one vector (numpy array) for one text."""
        return self.embed_many([text])[0]

    def embed_many(self, texts):
        """Return a list of vectors, one per text, in the same order."""
        raise NotImplementedError

embed(text)

Return one vector (numpy array) for one text.

Source code in src/rag_titan/embeddings/base_embedder.py

def embed(self, text):
    """Return one vector (numpy array) for one text."""
    return self.embed_many([text])[0]

embed_many(texts)

Return a list of vectors, one per text, in the same order.

Source code in src/rag_titan/embeddings/base_embedder.py

def embed_many(self, texts):
    """Return a list of vectors, one per text, in the same order."""
    raise NotImplementedError

FunctionEmbedder

Bases: BaseEmbedder

An embedder made from a function.

Parameters:

Name	Type	Description	Default
embed		function that takes a list of texts and returns one vector (list of numbers) per text, in the same order.	required
model		name to show for this model.	'function'

Source code in src/rag_titan/embeddings/function_embedder.py

class FunctionEmbedder(BaseEmbedder):
    """
    An embedder made from a function.

    Args:
        embed: function that takes a list of texts and returns one vector
            (list of numbers) per text, in the same order.
        model: name to show for this model.
    """

    def __init__(self, embed, model="function"):
        if not callable(embed):
            raise ValueError("embed must be a function: texts -> list of vectors")

        super().__init__(model)
        self.embed_function = embed

    def embed_many(self, texts):
        """Call the function and check it gave one vector per text."""
        if not texts:
            return []

        vectors = self.embed_function(list(texts))
        if not isinstance(vectors, list) or len(vectors) != len(texts):
            raise ValueError(f"The embed function must return {len(texts)} vectors, "
                             f"got {len(vectors) if isinstance(vectors, list) else vectors!r}")

        return [np.array(vector, dtype=np.float32) for vector in vectors]

embed_many(texts)

Call the function and check it gave one vector per text.

Source code in src/rag_titan/embeddings/function_embedder.py

def embed_many(self, texts):
    """Call the function and check it gave one vector per text."""
    if not texts:
        return []

    vectors = self.embed_function(list(texts))
    if not isinstance(vectors, list) or len(vectors) != len(texts):
        raise ValueError(f"The embed function must return {len(texts)} vectors, "
                         f"got {len(vectors) if isinstance(vectors, list) else vectors!r}")

    return [np.array(vector, dtype=np.float32) for vector in vectors]

HashEmbedder

Bases: BaseEmbedder

Turns text into word-count vectors of a fixed size (the "hashing trick").

Parameters:

Name	Type	Description	Default
dimensions		length of every vector; more means fewer words share a position.	512

Source code in src/rag_titan/embeddings/hash_embedder.py

class HashEmbedder(BaseEmbedder):
    """
    Turns text into word-count vectors of a fixed size (the "hashing trick").

    Args:
        dimensions: length of every vector; more means fewer words share a position.
    """

    def __init__(self, dimensions=512):
        if not isinstance(dimensions, int) or dimensions < 8:
            raise ValueError("dimensions must be a whole number of at least 8")

        super().__init__("hash")
        self.dimensions = dimensions

    def embed_many(self, texts):
        """Return one vector per text. The same text always gives the same vector."""
        return [self._embed_one(text) for text in texts]

    def _embed_one(self, text):
        """Add +1 or -1 at the hashed position of every word."""
        vector = np.zeros(self.dimensions, dtype=np.float32)

        for word in WORD.findall(text.lower()):
            # md5 instead of hash(): Python's hash() changes between runs
            number = int(hashlib.md5(word.encode("utf-8")).hexdigest(), 16)
            sign = 1.0 if number & 1 else -1.0   # signs make collisions cancel out
            vector[(number >> 1) % self.dimensions] += sign

        return vector

embed_many(texts)

Return one vector per text. The same text always gives the same vector.

Source code in src/rag_titan/embeddings/hash_embedder.py

def embed_many(self, texts):
    """Return one vector per text. The same text always gives the same vector."""
    return [self._embed_one(text) for text in texts]

OllamaEmbedder

Bases: BaseEmbedder

Creates embeddings with an Ollama server.

Parameters:

Name	Type	Description	Default
model		Ollama embedding model name.	'nomic-embed-text'
host		Ollama server address. None means http://localhost:11434.	None
batch_size		how many texts to send in one request.	32

Source code in src/rag_titan/embeddings/ollama_embedder.py

class OllamaEmbedder(BaseEmbedder):
    """
    Creates embeddings with an Ollama server.

    Args:
        model: Ollama embedding model name.
        host: Ollama server address. None means http://localhost:11434.
        batch_size: how many texts to send in one request.
    """

    def __init__(self, model="nomic-embed-text", host=None, batch_size=32):
        super().__init__(model)
        self.batch_size = batch_size
        self.host = parse_host(host)

    def embed_many(self, texts):
        """Return one numpy vector per text, in the same order."""
        vectors = []

        # send texts in small batches so large documents do not overload Ollama
        for start in range(0, len(texts), self.batch_size):
            batch = texts[start:start + self.batch_size]
            vectors.extend(self._embed_batch(batch))
            logger.debug("Embedded %d / %d texts", len(vectors), len(texts))

        return vectors

    def _embed_batch(self, batch):
        """Send one batch of texts to Ollama and return their vectors."""
        try:
            reply = post_json(self.host, "/api/embed", {"model": self.model, "input": batch})
        except RuntimeError as error:
            raise RuntimeError(
                f"Ollama embedding failed for model '{self.model}': {error}"
            ) from error

        vectors = reply.get("embeddings") if isinstance(reply, dict) else None
        # one vector per text, or the chunks and vectors would get out of step
        if not isinstance(vectors, list) or len(vectors) != len(batch):
            raise RuntimeError(
                f"Ollama sent {len(vectors or [])} embeddings for {len(batch)} texts "
                f"(model '{self.model}')"
            )

        return [np.array(vector, dtype=np.float32) for vector in vectors]

embed_many(texts)

Return one numpy vector per text, in the same order.

Source code in src/rag_titan/embeddings/ollama_embedder.py

def embed_many(self, texts):
    """Return one numpy vector per text, in the same order."""
    vectors = []

    # send texts in small batches so large documents do not overload Ollama
    for start in range(0, len(texts), self.batch_size):
        batch = texts[start:start + self.batch_size]
        vectors.extend(self._embed_batch(batch))
        logger.debug("Embedded %d / %d texts", len(vectors), len(texts))

    return vectors

RecordingEmbedder

Bases: BaseEmbedder

Passes every text to a real embedder and remembers its vector.

Parameters:

Name	Type	Description	Default
embedder		any BaseEmbedder, for example an OllamaEmbedder.	required
decimals		digits kept per number in vectors (smaller files).	5

Source code in src/rag_titan/embeddings/replay_embedder.py

class RecordingEmbedder(BaseEmbedder):
    """
    Passes every text to a real embedder and remembers its vector.

    Args:
        embedder: any BaseEmbedder, for example an OllamaEmbedder.
        decimals: digits kept per number in `vectors` (smaller files).
    """

    def __init__(self, embedder, decimals=5):
        super().__init__(embedder.model)
        self.embedder = embedder
        self.decimals = decimals
        self.vectors = {}

    def embed_many(self, texts):
        """Embed new texts with the real model, save every vector, and return the saved ones."""
        new = [text for text in texts if text_key(text) not in self.vectors]
        if new:
            for text, vector in zip(new, self.embedder.embed_many(new)):
                self.vectors[text_key(text)] = [round(float(x), self.decimals) for x in vector]

        # return the rounded vectors, so a replay later gives exactly the same scores
        return [np.array(self.vectors[text_key(text)], dtype=np.float32) for text in texts]

embed_many(texts)

Embed new texts with the real model, save every vector, and return the saved ones.

Source code in src/rag_titan/embeddings/replay_embedder.py

def embed_many(self, texts):
    """Embed new texts with the real model, save every vector, and return the saved ones."""
    new = [text for text in texts if text_key(text) not in self.vectors]
    if new:
        for text, vector in zip(new, self.embedder.embed_many(new)):
            self.vectors[text_key(text)] = [round(float(x), self.decimals) for x in vector]

    # return the rounded vectors, so a replay later gives exactly the same scores
    return [np.array(self.vectors[text_key(text)], dtype=np.float32) for text in texts]

ReplayEmbedder

Bases: BaseEmbedder

Returns recorded vectors.

Parameters:

Name	Type	Description	Default
vectors		dict of {text key: list of numbers}, e.g. RecordingEmbedder.vectors.	required
model		name to show, e.g. the model the vectors came from.	'replay'

Source code in src/rag_titan/embeddings/replay_embedder.py

class ReplayEmbedder(BaseEmbedder):
    """
    Returns recorded vectors.

    Args:
        vectors: dict of {text key: list of numbers}, e.g. RecordingEmbedder.vectors.
        model: name to show, e.g. the model the vectors came from.
    """

    def __init__(self, vectors, model="replay"):
        if not isinstance(vectors, dict):
            raise ValueError("vectors must be a dict of {text key: list of numbers}")

        super().__init__(model)
        self.vectors = vectors

    def embed_many(self, texts):
        """Return the recorded vector for every text, in the same order."""
        missing = [text for text in texts if text_key(text) not in self.vectors]
        if missing:
            raise ValueError(
                f"No recorded vector for {len(missing)} text(s), e.g. {missing[0][:60]!r}. "
                "The documents, chunker settings or question differ from the recording."
            )

        return [np.array(self.vectors[text_key(text)], dtype=np.float32) for text in texts]

embed_many(texts)

Return the recorded vector for every text, in the same order.

Source code in src/rag_titan/embeddings/replay_embedder.py

def embed_many(self, texts):
    """Return the recorded vector for every text, in the same order."""
    missing = [text for text in texts if text_key(text) not in self.vectors]
    if missing:
        raise ValueError(
            f"No recorded vector for {len(missing)} text(s), e.g. {missing[0][:60]!r}. "
            "The documents, chunker settings or question differ from the recording."
        )

    return [np.array(self.vectors[text_key(text)], dtype=np.float32) for text in texts]

Vector stores that save chunks and search them by similarity.

BaseStore

Saves chunks with their vectors and finds the most similar ones.

Source code in src/rag_titan/vector_stores/base_store.py

class BaseStore:
    """Saves chunks with their vectors and finds the most similar ones."""

    def add(self, chunks, vectors):
        """Save chunks together with their vectors (same order)."""
        raise NotImplementedError

    def search(self, query_vector, top_k=5):
        """
        Return the top_k most similar chunks, best first.
        Each result is the chunk dict plus a "score" key.
        """
        raise NotImplementedError

    def clear(self):
        """Remove everything from the store."""
        raise NotImplementedError

    def count(self):
        """Return how many chunks are stored."""
        raise NotImplementedError

add(chunks, vectors)

Save chunks together with their vectors (same order).

Source code in src/rag_titan/vector_stores/base_store.py

def add(self, chunks, vectors):
    """Save chunks together with their vectors (same order)."""
    raise NotImplementedError

clear()

Remove everything from the store.

Source code in src/rag_titan/vector_stores/base_store.py

def clear(self):
    """Remove everything from the store."""
    raise NotImplementedError

count()

Return how many chunks are stored.

Source code in src/rag_titan/vector_stores/base_store.py

def count(self):
    """Return how many chunks are stored."""
    raise NotImplementedError

search(query_vector, top_k=5)

Return the top_k most similar chunks, best first. Each result is the chunk dict plus a "score" key.

Source code in src/rag_titan/vector_stores/base_store.py

def search(self, query_vector, top_k=5):
    """
    Return the top_k most similar chunks, best first.
    Each result is the chunk dict plus a "score" key.
    """
    raise NotImplementedError

MemoryStore

Bases: BaseStore

Keeps everything in a Python list. Good for learning and small data.

Source code in src/rag_titan/vector_stores/memory_store.py

class MemoryStore(BaseStore):
    """Keeps everything in a Python list. Good for learning and small data."""

    def __init__(self):
        self.records = []

    def add(self, chunks, vectors):
        """Save each chunk with its vector."""
        if len(chunks) != len(vectors):
            raise ValueError("chunks and vectors must have the same length")

        for chunk, vector in zip(chunks, vectors):
            self.records.append({**chunk, "vector": vector})

    def search(self, query_vector, top_k=5):
        """Compare the query with every chunk and return the best top_k."""
        matches = []

        for record in self.records:
            # copy the chunk without its vector, then add the similarity score
            match = {key: value for key, value in record.items() if key != "vector"}
            match["score"] = self._cosine_similarity(query_vector, record["vector"])
            matches.append(match)

        matches.sort(key=lambda match: match["score"], reverse=True)
        return matches[:top_k]

    def clear(self):
        """Remove every stored chunk."""
        self.records = []

    def count(self):
        """Return how many chunks are stored."""
        return len(self.records)

    def _cosine_similarity(self, vector1, vector2):
        """Return a number from -1 to 1. Higher means more similar."""
        denominator = np.linalg.norm(vector1) * np.linalg.norm(vector2)

        if denominator == 0:
            return 0.0

        return float(np.dot(vector1, vector2) / denominator)

add(chunks, vectors)

Save each chunk with its vector.

Source code in src/rag_titan/vector_stores/memory_store.py

def add(self, chunks, vectors):
    """Save each chunk with its vector."""
    if len(chunks) != len(vectors):
        raise ValueError("chunks and vectors must have the same length")

    for chunk, vector in zip(chunks, vectors):
        self.records.append({**chunk, "vector": vector})

clear()

Remove every stored chunk.

Source code in src/rag_titan/vector_stores/memory_store.py

def clear(self):
    """Remove every stored chunk."""
    self.records = []

count()

Return how many chunks are stored.

Source code in src/rag_titan/vector_stores/memory_store.py

def count(self):
    """Return how many chunks are stored."""
    return len(self.records)

search(query_vector, top_k=5)

Compare the query with every chunk and return the best top_k.

Source code in src/rag_titan/vector_stores/memory_store.py

def search(self, query_vector, top_k=5):
    """Compare the query with every chunk and return the best top_k."""
    matches = []

    for record in self.records:
        # copy the chunk without its vector, then add the similarity score
        match = {key: value for key, value in record.items() if key != "vector"}
        match["score"] = self._cosine_similarity(query_vector, record["vector"])
        matches.append(match)

    matches.sort(key=lambda match: match["score"], reverse=True)
    return matches[:top_k]

Retrieval: find similar chunks, then rerank them.

BaseRetriever

Finds the chunks that best match a question.

Every result is a chunk dict ("id", "source", "text", offsets, ...) plus: "score": the retriever's score, higher is better. "retriever": the name of the retriever that found it, e.g. "dense". "score_trace": how the score was made, e.g. {"dense": 0.83}. A hybrid retriever would add one entry per part, e.g. {"bm25": 7.1, "dense": 0.83, "rrf": 0.032}.

"id" and "source" are the chunk id and source id of the evidence.

Source code in src/rag_titan/retrieval/base_retriever.py

class BaseRetriever:
    """
    Finds the chunks that best match a question.

    Every result is a chunk dict ("id", "source", "text", offsets, ...) plus:
        "score": the retriever's score, higher is better.
        "retriever": the `name` of the retriever that found it, e.g. "dense".
        "score_trace": how the score was made, e.g. {"dense": 0.83}.
            A hybrid retriever would add one entry per part, e.g.
            {"bm25": 7.1, "dense": 0.83, "rrf": 0.032}.

    "id" and "source" are the chunk id and source id of the evidence.
    """

    name = "base"

    def retrieve(self, query, top_k=5):
        """Return the top_k best chunks, best first."""
        raise NotImplementedError

    def has_documents(self):
        """Return True when there is at least one chunk to search."""
        raise NotImplementedError

    def _check_top_k(self, top_k):
        """Raise a clear error for a top_k that cannot return anything."""
        if not isinstance(top_k, int) or isinstance(top_k, bool) or top_k < 1:
            raise ValueError(f"top_k must be a whole number of at least 1, got {top_k!r}")

    def _with_trace(self, matches):
        """Return copies of the matches with "retriever" and "score_trace" added."""
        traced = []

        for match in matches:
            new_match = dict(match)
            new_match["retriever"] = self.name
            new_match["score_trace"] = {self.name: match["score"]}
            traced.append(new_match)

        return traced

has_documents()

Return True when there is at least one chunk to search.

Source code in src/rag_titan/retrieval/base_retriever.py

def has_documents(self):
    """Return True when there is at least one chunk to search."""
    raise NotImplementedError

retrieve(query, top_k=5)

Return the top_k best chunks, best first.

Source code in src/rag_titan/retrieval/base_retriever.py

def retrieve(self, query, top_k=5):
    """Return the top_k best chunks, best first."""
    raise NotImplementedError

DenseRetriever

Bases: BaseRetriever

Finds chunks whose vectors are similar to the question's vector.

Parameters:

Name	Type	Description	Default
embedder		a BaseEmbedder (must be the same one used for indexing).	required
store		a BaseStore that holds the indexed chunks.	required
dedupe_overlap		drop a result that overlaps a better result from the same file by at least this share of the shorter chunk (0.5 = half). None keeps every result, like before 0.5.0.	0.5

Source code in src/rag_titan/retrieval/dense_retriever.py

class DenseRetriever(BaseRetriever):
    """
    Finds chunks whose vectors are similar to the question's vector.

    Args:
        embedder: a BaseEmbedder (must be the same one used for indexing).
        store: a BaseStore that holds the indexed chunks.
        dedupe_overlap: drop a result that overlaps a better result from the
            same file by at least this share of the shorter chunk (0.5 = half).
            None keeps every result, like before 0.5.0.
    """

    name = "dense"

    def __init__(self, embedder, store, dedupe_overlap=0.5):
        if dedupe_overlap is not None and not 0 < dedupe_overlap <= 1:
            raise ValueError("dedupe_overlap must be more than 0 and at most 1, or None")

        self.embedder = embedder
        self.store = store
        self.dedupe_overlap = dedupe_overlap

    def has_documents(self):
        """Return True when the store holds at least one chunk."""
        return self.store.count() > 0

    def retrieve(self, query, top_k=5):
        """
        Return the top_k most similar chunks, best first.

        Each result is a chunk dict with "score", "retriever" and "score_trace"
        keys (see BaseRetriever). Near-copies of a better result are removed
        (see dedupe_overlap), so twice as many candidates are fetched to still
        return top_k different chunks.
        """
        self._check_top_k(top_k)

        if not self.has_documents():
            logger.warning("Vector store is empty, nothing to retrieve")
            return []

        query_vector = self.embedder.embed(query)
        if self.dedupe_overlap is None:
            matches = self.store.search(query_vector, top_k=top_k)
        else:
            candidates = self.store.search(query_vector, top_k=2 * top_k)
            different = remove_duplicates(candidates, self.dedupe_overlap)
            logger.debug("Removed %d overlapping results", len(candidates) - len(different))
            matches = different[:top_k]

        for match in matches:
            logger.debug("Retrieved %s (score %.4f)", match["id"], match["score"])

        return self._with_trace(matches)

has_documents()

Return True when the store holds at least one chunk.

Source code in src/rag_titan/retrieval/dense_retriever.py

def has_documents(self):
    """Return True when the store holds at least one chunk."""
    return self.store.count() > 0

retrieve(query, top_k=5)

Return the top_k most similar chunks, best first.

Each result is a chunk dict with "score", "retriever" and "score_trace" keys (see BaseRetriever). Near-copies of a better result are removed (see dedupe_overlap), so twice as many candidates are fetched to still return top_k different chunks.

Source code in src/rag_titan/retrieval/dense_retriever.py

def retrieve(self, query, top_k=5):
    """
    Return the top_k most similar chunks, best first.

    Each result is a chunk dict with "score", "retriever" and "score_trace"
    keys (see BaseRetriever). Near-copies of a better result are removed
    (see dedupe_overlap), so twice as many candidates are fetched to still
    return top_k different chunks.
    """
    self._check_top_k(top_k)

    if not self.has_documents():
        logger.warning("Vector store is empty, nothing to retrieve")
        return []

    query_vector = self.embedder.embed(query)
    if self.dedupe_overlap is None:
        matches = self.store.search(query_vector, top_k=top_k)
    else:
        candidates = self.store.search(query_vector, top_k=2 * top_k)
        different = remove_duplicates(candidates, self.dedupe_overlap)
        logger.debug("Removed %d overlapping results", len(candidates) - len(different))
        matches = different[:top_k]

    for match in matches:
        logger.debug("Retrieved %s (score %.4f)", match["id"], match["score"])

    return self._with_trace(matches)

overlap_fraction(a, b)

How much two chunks overlap, from 0.0 (not at all) to 1.0.

Measured as shared characters divided by the length of the shorter chunk, so a chunk fully inside another counts as 1.0. Chunks from different files, or without offsets, give 0.0.

Source code in src/rag_titan/retrieval/dedupe.py

def overlap_fraction(a, b):
    """
    How much two chunks overlap, from 0.0 (not at all) to 1.0.

    Measured as shared characters divided by the length of the shorter chunk,
    so a chunk fully inside another counts as 1.0. Chunks from different files,
    or without offsets, give 0.0.
    """
    if a.get("source") != b.get("source") or not (has_offsets(a) and has_offsets(b)):
        return 0.0

    shared = min(a["end_offset"], b["end_offset"]) - max(a["start_offset"], b["start_offset"])
    shorter = min(a["end_offset"] - a["start_offset"], b["end_offset"] - b["start_offset"])
    return max(shared, 0) / shorter if shorter > 0 else 0.0

remove_duplicates(matches, min_overlap=0.5)

Keep matches best-first and drop the ones that repeat a better match.

Parameters:

Name	Type	Description	Default
matches		search results, best first.	required
min_overlap		drop a match that overlaps a kept one by at least this share of the shorter chunk (0.5 = half of it).	0.5

Returns:

Type	Description
	The matches that are left, in the same order.

Source code in src/rag_titan/retrieval/dedupe.py

def remove_duplicates(matches, min_overlap=0.5):
    """
    Keep matches best-first and drop the ones that repeat a better match.

    Args:
        matches: search results, best first.
        min_overlap: drop a match that overlaps a kept one by at least this
            share of the shorter chunk (0.5 = half of it).

    Returns:
        The matches that are left, in the same order.
    """
    if not 0 < min_overlap <= 1:
        raise ValueError("min_overlap must be more than 0 and at most 1")

    kept = []
    for match in matches:
        if not is_duplicate(match, kept, min_overlap):
            kept.append(match)
    return kept

Rerankers that re-order retrieved chunks.

BaseReranker

Re-orders retrieved chunks so the most useful ones come first.

Source code in src/rag_titan/retrieval/rerankers/base_reranker.py

class BaseReranker:
    """Re-orders retrieved chunks so the most useful ones come first."""

    def rerank(self, query, matches, top_n=3):
        """
        Score every match again and keep the best ones.

        Args:
            query: the user's question.
            matches: list of dicts from the retriever (each has "score").
            top_n: how many matches to keep.

        Returns:
            The best top_n matches, each with a "rerank_score" key.
        """
        reranked = []

        for match in matches:
            # copy so the retriever's results are never changed
            new_match = dict(match)
            new_match["rerank_score"] = self.score(query, match)
            reranked.append(new_match)

        reranked.sort(key=lambda match: match["rerank_score"], reverse=True)
        return reranked[:top_n]

    def score(self, query, match):
        """Return a number. Higher means more relevant."""
        raise NotImplementedError

rerank(query, matches, top_n=3)

Score every match again and keep the best ones.

Parameters:

Name	Type	Description	Default
query		the user's question.	required
matches		list of dicts from the retriever (each has "score").	required
top_n		how many matches to keep.	3

Returns:

Type	Description
	The best top_n matches, each with a "rerank_score" key.

Source code in src/rag_titan/retrieval/rerankers/base_reranker.py

def rerank(self, query, matches, top_n=3):
    """
    Score every match again and keep the best ones.

    Args:
        query: the user's question.
        matches: list of dicts from the retriever (each has "score").
        top_n: how many matches to keep.

    Returns:
        The best top_n matches, each with a "rerank_score" key.
    """
    reranked = []

    for match in matches:
        # copy so the retriever's results are never changed
        new_match = dict(match)
        new_match["rerank_score"] = self.score(query, match)
        reranked.append(new_match)

    reranked.sort(key=lambda match: match["rerank_score"], reverse=True)
    return reranked[:top_n]

score(query, match)

Return a number. Higher means more relevant.

Source code in src/rag_titan/retrieval/rerankers/base_reranker.py

def score(self, query, match):
    """Return a number. Higher means more relevant."""
    raise NotImplementedError

KeywordReranker

Bases: BaseReranker

Adds a small bonus for every word shared by the question and the chunk. final score = semantic score + keyword_weight * shared words

Parameters:

Name	Type	Description	Default
keyword_weight		bonus added for each shared word.	0.05

Source code in src/rag_titan/retrieval/rerankers/keyword_reranker.py

class KeywordReranker(BaseReranker):
    """
    Adds a small bonus for every word shared by the question and the chunk.
    final score = semantic score + keyword_weight * shared words

    Args:
        keyword_weight: bonus added for each shared word.
    """

    def __init__(self, keyword_weight=0.05):
        self.keyword_weight = keyword_weight

    def score(self, query, match):
        """Return the semantic score plus the keyword bonus."""
        shared_words = self._words(query) & self._words(match["text"])
        return match["score"] + self.keyword_weight * len(shared_words)

    def _words(self, text):
        """Return the set of lowercase words in the text."""
        return set(text.lower().translate(PUNCTUATION).split())

score(query, match)

Return the semantic score plus the keyword bonus.

Source code in src/rag_titan/retrieval/rerankers/keyword_reranker.py

def score(self, query, match):
    """Return the semantic score plus the keyword bonus."""
    shared_words = self._words(query) & self._words(match["text"])
    return match["score"] + self.keyword_weight * len(shared_words)

NoReranker

Bases: BaseReranker

Keeps the retriever's order. Useful to compare against real rerankers.

Source code in src/rag_titan/retrieval/rerankers/no_reranker.py

class NoReranker(BaseReranker):
    """Keeps the retriever's order. Useful to compare against real rerankers."""

    def score(self, query, match):
        """Return the retriever's own score unchanged."""
        return match["score"]

score(query, match)

Return the retriever's own score unchanged.

Source code in src/rag_titan/retrieval/rerankers/no_reranker.py

def score(self, query, match):
    """Return the retriever's own score unchanged."""
    return match["score"]

Evaluation: measure how well a RAG system works.

Two ways to use it

evaluate_rag(), compare_setups(), evaluate_retriever() run rag-titan pipelines on test questions and score them. RAGEvaluator and the stage evaluators (ChunkEvaluator, RetrievalEvaluator, ...) score the logged outputs of any RAG system: data in, report out.

AugmentationEvaluator

Scores the chunks that were packed into the prompt for each query.

Source code in src/rag_titan/evaluation/stages/augmentation.py

class AugmentationEvaluator:
    """Scores the chunks that were packed into the prompt for each query."""

    def evaluate(self, chunks, queries, document_text=None, documents=None,
                 max_context_tokens=None):
        """
        Score every query that has "context_ids".

        Args:
            chunks: all chunks: texts, or dicts with "id", "text", "start", "end", "doc_id".
            queries: list of dicts with "id", "context_ids" and the gold labels:
                "expected_evidence" (needs the source text) and/or
                "expected_chunk_ids". "max_context_tokens" can be set per query.
            document_text: the source text of a single document.
            documents: {doc_id: text} when there are several documents.
            max_context_tokens: the context budget in words, for every query.

        Returns:
            {"summary": the mean of each metric, "per_query": one dict per
            query, "warnings": chunks that could not be placed in the text}.
        """
        check_queries(queries)
        documents = as_documents(document_text, documents)
        items, warnings = normalize_chunks(chunks, documents)
        by_id = {item["id"]: item for item in items}
        tokens = document_tokens(documents)

        per_query = [self._score_query(query, by_id, documents, tokens, max_context_tokens)
                     for query in queries if "context_ids" in query]

        summary = {"query_count": len(per_query)}
        for key in SUMMARY_KEYS:
            summary[key] = average([scores[key] for scores in per_query])
        return {"summary": summary, "per_query": per_query, "warnings": warnings}

    def _score_query(self, query, by_id, documents, tokens, max_context_tokens):
        """Return every context score of one query."""
        selected = ranked_ids(query, "context_ids")
        used = [by_id[chunk_id] for chunk_id in selected if chunk_id in by_id]
        evidence = []
        if documents:
            evidence = evidence_spans(query.get("expected_evidence"), documents)
        budget = query.get("max_context_tokens", max_context_tokens)

        scores = {"query_id": query["id"]}
        scores.update(self._evidence_scores(used, evidence, tokens))
        scores.update(self._budget_scores(used, tokens, documents, budget))
        scores["evidence_edge_placement"] = self._edge_placement(used, evidence, query)
        scores["missing_chunk_ids"] = [cid for cid in selected if cid not in by_id]
        return scores

    def _evidence_scores(self, used, evidence, tokens):
        """
        Compare the context words (C) with the gold evidence words (G).

            Context Precision  = |C ∩ G| / |C|
            Context Recall     = |C ∩ G| / |G|
            Context Noise      = 1 - Context Precision
            Context Sufficient = 1 when every gold span is fully inside a context chunk
        """
        if not evidence:
            return {key: None for key in SUMMARY_KEYS[:4]}

        spans = chunk_spans(used)
        context, gold = covered_tokens(tokens, spans), covered_tokens(tokens, evidence)
        precision = len(context & gold) / len(context) if context else None
        return {
            "context_precision": precision,
            "context_recall": len(context & gold) / len(gold) if gold else None,
            "context_noise": None if precision is None else 1 - precision,
            "context_sufficient": float(all(is_inside(span, spans) for span in evidence)),
        }

    def _budget_scores(self, used, tokens, documents, budget):
        """
        Size of the context in words, repeated words, and the budget check.

            Duplicate Ratio  = (context words - distinct source words) / context words
            Budget Compliant = 1 when context words <= max_context_tokens
        """
        count = sum(len(tokenize(item["text"])) for item in used)
        scores = {"context_tokens": count, "duplicate_ratio": None, "budget_compliant": None}

        placed = documents and all(item["start"] is not None for item in used)
        if count and placed:
            distinct = len(covered_tokens(tokens, chunk_spans(used)))
            scores["duplicate_ratio"] = max(0.0, (count - distinct) / count)
        if budget is not None:
            scores["budget_compliant"] = float(count <= budget)
        return scores

    def _edge_placement(self, used, evidence, query):
        """
        Is the evidence at the start or end of the context (1.0) or in the middle (0.0)?
        LLMs often overlook text in the middle of a long prompt ("lost in the middle").
        """
        gold_ids = {cid for cid, grade in relevance_of(query).items() if grade > 0}
        placements = []
        for number, item in enumerate(used):
            touches = item["start"] is not None and any(
                doc_id == item["doc_id"] and item["start"] < end and item["end"] > start
                for doc_id, start, end in evidence)
            if item["id"] in gold_ids or touches:
                position = number / (len(used) - 1) if len(used) > 1 else 0.0
                placements.append(abs(2 * position - 1))
        return average(placements)

evaluate(chunks, queries, document_text=None, documents=None, max_context_tokens=None)

Score every query that has "context_ids".

Parameters:

Name	Type	Description	Default
chunks		all chunks: texts, or dicts with "id", "text", "start", "end", "doc_id".	required
queries		list of dicts with "id", "context_ids" and the gold labels: "expected_evidence" (needs the source text) and/or "expected_chunk_ids". "max_context_tokens" can be set per query.	required
document_text		the source text of a single document.	None
documents		{doc_id: text} when there are several documents.	None
max_context_tokens		the context budget in words, for every query.	None

Returns:

Type	Description
	{"summary": the mean of each metric, "per_query": one dict per
	query, "warnings": chunks that could not be placed in the text}.

Source code in src/rag_titan/evaluation/stages/augmentation.py

def evaluate(self, chunks, queries, document_text=None, documents=None,
             max_context_tokens=None):
    """
    Score every query that has "context_ids".

    Args:
        chunks: all chunks: texts, or dicts with "id", "text", "start", "end", "doc_id".
        queries: list of dicts with "id", "context_ids" and the gold labels:
            "expected_evidence" (needs the source text) and/or
            "expected_chunk_ids". "max_context_tokens" can be set per query.
        document_text: the source text of a single document.
        documents: {doc_id: text} when there are several documents.
        max_context_tokens: the context budget in words, for every query.

    Returns:
        {"summary": the mean of each metric, "per_query": one dict per
        query, "warnings": chunks that could not be placed in the text}.
    """
    check_queries(queries)
    documents = as_documents(document_text, documents)
    items, warnings = normalize_chunks(chunks, documents)
    by_id = {item["id"]: item for item in items}
    tokens = document_tokens(documents)

    per_query = [self._score_query(query, by_id, documents, tokens, max_context_tokens)
                 for query in queries if "context_ids" in query]

    summary = {"query_count": len(per_query)}
    for key in SUMMARY_KEYS:
        summary[key] = average([scores[key] for scores in per_query])
    return {"summary": summary, "per_query": per_query, "warnings": warnings}

ChunkEvaluator

Scores the output of a chunker against the source text and gold labels.

Source code in src/rag_titan/evaluation/stages/chunking.py

class ChunkEvaluator:
    """Scores the output of a chunker against the source text and gold labels."""

    def evaluate(self, chunks, document_text=None, documents=None, expected_chunks=None,
                 expected_evidence=None, structural_blocks=None, required_metadata=None,
                 expected_metadata=None, size_limits=None, boundary_tolerance=0,
                 sentence_vectors=None, chunk_vectors=None):
        """
        Score the chunks.

        Args:
            chunks: texts, or dicts with "id", "text", "start", "end", "doc_id", "metadata".
            document_text: the source text of a single document.
            documents: {doc_id: text} when there are several documents.
            expected_chunks: the ideal chunks marked by a human (for boundary F1).
            expected_evidence: gold quotes or {"start", "end"} spans that must stay whole.
            structural_blocks: tables, lists or code that must stay whole (same format).
            required_metadata: keys every chunk must have, e.g. ["source", "page"].
            expected_metadata: {chunk_id: {key: value}} of correct metadata values.
            size_limits: (min, max) words of a valid chunk.
            boundary_tolerance: characters two boundaries may be apart and still match.
            sentence_vectors: {chunk_id: [vector per sentence]} for semantic cohesion.
            chunk_vectors: {chunk_id: vector} for semantic chunk stickiness.

        Returns:
            A dict of metric name -> value (None = not measured) and "warnings".
        """
        check_size_limits(size_limits)
        documents = as_documents(document_text, documents)
        items, warnings = normalize_chunks(chunks, documents)

        scores = {"chunk_count": len(items)}
        scores.update(coverage_scores(items, documents))
        scores.update(size_scores(items, size_limits))
        scores.update(self._gold_scores(items, documents, expected_evidence, structural_blocks))
        scores["metadata_completeness"] = metadata_completeness(items, required_metadata)
        scores["metadata_accuracy"] = metadata_accuracy(items, expected_metadata)
        scores.update(lexical_scores(items))
        scores.update(semantic_scores(items, sentence_vectors, chunk_vectors))
        scores["boundary_f1"] = self._boundary_f1(items, expected_chunks, documents,
                                                  boundary_tolerance)
        scores["warnings"] = warnings
        return scores

    def _gold_scores(self, items, documents, expected_evidence, structural_blocks):
        """Check that gold evidence and structural blocks were not cut by a boundary."""
        self._need_documents(documents, expected_evidence, "expected_evidence")
        self._need_documents(documents, structural_blocks, "structural_blocks")

        evidence = evidence_spans(expected_evidence, documents)
        blocks = evidence_spans(structural_blocks, documents)
        return {"evidence_completeness": intact_rate(evidence, items),
                "evidence_fragmentation": fragmentation(evidence, items),
                "structural_integrity": intact_rate(blocks, items)}

    def _boundary_f1(self, items, expected_chunks, documents, tolerance):
        """Place the ideal chunks in the text and compare the split points."""
        if expected_chunks is None:
            return None
        self._need_documents(documents, expected_chunks, "expected_chunks")
        expected_items, _ = normalize_chunks(expected_chunks, documents)
        return boundary_f1(items, expected_items, tolerance)

    def _need_documents(self, documents, value, name):
        """Gold spans are places in the source text, so the text must be given."""
        if value and not documents:
            raise ValueError(f"{name} needs the source text: give document_text or documents")

evaluate(chunks, document_text=None, documents=None, expected_chunks=None, expected_evidence=None, structural_blocks=None, required_metadata=None, expected_metadata=None, size_limits=None, boundary_tolerance=0, sentence_vectors=None, chunk_vectors=None)

Score the chunks.

Parameters:

Name	Type	Description	Default
chunks		texts, or dicts with "id", "text", "start", "end", "doc_id", "metadata".	required
document_text		the source text of a single document.	None
documents		{doc_id: text} when there are several documents.	None
expected_chunks		the ideal chunks marked by a human (for boundary F1).	None
expected_evidence		gold quotes or {"start", "end"} spans that must stay whole.	None
structural_blocks		tables, lists or code that must stay whole (same format).	None
required_metadata		keys every chunk must have, e.g. ["source", "page"].	None
expected_metadata		{chunk_id: {key: value}} of correct metadata values.	None
size_limits		(min, max) words of a valid chunk.	None
boundary_tolerance		characters two boundaries may be apart and still match.	0
sentence_vectors		{chunk_id: [vector per sentence]} for semantic cohesion.	None
chunk_vectors		{chunk_id: vector} for semantic chunk stickiness.	None

Returns:

Type	Description
	A dict of metric name -> value (None = not measured) and "warnings".

Source code in src/rag_titan/evaluation/stages/chunking.py

def evaluate(self, chunks, document_text=None, documents=None, expected_chunks=None,
             expected_evidence=None, structural_blocks=None, required_metadata=None,
             expected_metadata=None, size_limits=None, boundary_tolerance=0,
             sentence_vectors=None, chunk_vectors=None):
    """
    Score the chunks.

    Args:
        chunks: texts, or dicts with "id", "text", "start", "end", "doc_id", "metadata".
        document_text: the source text of a single document.
        documents: {doc_id: text} when there are several documents.
        expected_chunks: the ideal chunks marked by a human (for boundary F1).
        expected_evidence: gold quotes or {"start", "end"} spans that must stay whole.
        structural_blocks: tables, lists or code that must stay whole (same format).
        required_metadata: keys every chunk must have, e.g. ["source", "page"].
        expected_metadata: {chunk_id: {key: value}} of correct metadata values.
        size_limits: (min, max) words of a valid chunk.
        boundary_tolerance: characters two boundaries may be apart and still match.
        sentence_vectors: {chunk_id: [vector per sentence]} for semantic cohesion.
        chunk_vectors: {chunk_id: vector} for semantic chunk stickiness.

    Returns:
        A dict of metric name -> value (None = not measured) and "warnings".
    """
    check_size_limits(size_limits)
    documents = as_documents(document_text, documents)
    items, warnings = normalize_chunks(chunks, documents)

    scores = {"chunk_count": len(items)}
    scores.update(coverage_scores(items, documents))
    scores.update(size_scores(items, size_limits))
    scores.update(self._gold_scores(items, documents, expected_evidence, structural_blocks))
    scores["metadata_completeness"] = metadata_completeness(items, required_metadata)
    scores["metadata_accuracy"] = metadata_accuracy(items, expected_metadata)
    scores.update(lexical_scores(items))
    scores.update(semantic_scores(items, sentence_vectors, chunk_vectors))
    scores["boundary_f1"] = self._boundary_f1(items, expected_chunks, documents,
                                              boundary_tolerance)
    scores["warnings"] = warnings
    return scores

EmbeddingEvaluator

Scores chunk and query vectors, and the results of a vector index.

Source code in src/rag_titan/evaluation/stages/embedding.py

class EmbeddingEvaluator:
    """Scores chunk and query vectors, and the results of a vector index."""

    def evaluate(self, vectors, query_vectors=None, relevant_ids=None, indexed_ids=None,
                 index_results=None, k=5):
        """
        Score the vectors.

        Args:
            vectors: {chunk_id: vector} of every chunk.
            query_vectors: {query_id: vector}.
            relevant_ids: {query_id: [gold chunk ids]}.
            indexed_ids: the chunk ids really stored in your vector index.
            index_results: {query_id: [chunk ids the index returned, best first]}.
            k: how many neighbours to compare.

        Returns:
            A dict of metric name -> value (None = not measured).
        """
        check_k(k)
        if not isinstance(vectors, dict):
            raise ValueError("vectors must be a dict of {chunk_id: vector}")

        scores = self._health(vectors)
        scores["index_coverage"] = None
        if indexed_ids is not None and vectors:
            scores["index_coverage"] = len(set(vectors) & set(indexed_ids)) / len(vectors)

        scores.update(self._search_scores(vectors, query_vectors or {}, relevant_ids or {},
                                          index_results, k))
        return scores

    def _health(self, vectors):
        """
        Basic checks on the stored vectors.

            zero_norm_rate         vectors of length 0 / all vectors (cannot be searched)
            duplicate_vector_rate  vectors equal to an earlier one / all vectors
        """
        sizes = {len(vector) for vector in vectors.values()}
        if len(sizes) > 1:
            raise ValueError(f"Vectors have different sizes: {sorted(sizes)}")

        norms = []
        for chunk_id, vector in vectors.items():
            if not all(is_number(value) for value in vector):
                raise ValueError(f"Vector of {chunk_id!r} has a value that is not a finite number")
            norms.append(math.sqrt(sum(value * value for value in vector)))

        count = len(vectors)
        distinct = len({tuple(vector) for vector in vectors.values()})
        return {
            "vector_count": count,
            "dimension": next(iter(sizes), None),
            "zero_norm_rate": sum(norm == 0 for norm in norms) / count if count else None,
            "duplicate_vector_rate": (count - distinct) / count if count else None,
            "mean_vector_norm": average(norms),
        }

    def _search_scores(self, vectors, query_vectors, relevant_ids, index_results, k):
        """
        Search every query by exact cosine and compare with the gold chunks and the index.

            query_topk_recall    gold chunks in the exact top K / gold chunks
            ann_recall_at_k      exact top K also returned by your index / exact top K
            positive_similarity  mean cos(query, gold chunk)
            negative_similarity  mean cos(query, other chunk)
            similarity_margin    positive - negative (higher = easier to separate)
        """
        recalls, ann_recalls, positive, negative = [], [], [], []
        for query_id, query_vector in query_vectors.items():
            scored = [(chunk_id, cosine(query_vector, vector))
                      for chunk_id, vector in vectors.items()]
            scored = [(chunk_id, score) for chunk_id, score in scored if score is not None]
            exact = [chunk_id for chunk_id, _ in
                     sorted(scored, key=lambda pair: pair[1], reverse=True)[:k]]

            gold = set(relevant_ids.get(query_id) or [])
            if gold:
                recalls.append(len(set(exact) & gold) / len(gold))
                positive += [score for chunk_id, score in scored if chunk_id in gold]
                negative += [score for chunk_id, score in scored if chunk_id not in gold]
            if index_results is not None and query_id in index_results and exact:
                returned = set(index_results[query_id][:k])
                ann_recalls.append(len(returned & set(exact)) / len(exact))

        return {
            "query_topk_recall": average(recalls),
            "ann_recall_at_k": average(ann_recalls),
            "positive_similarity": average(positive),
            "negative_similarity": average(negative),
            "similarity_margin": difference(average(positive), average(negative)),
        }

evaluate(vectors, query_vectors=None, relevant_ids=None, indexed_ids=None, index_results=None, k=5)

Score the vectors.

Parameters:

Name	Type	Description	Default
vectors		{chunk_id: vector} of every chunk.	required
query_vectors		{query_id: vector}.	None
relevant_ids		{query_id: [gold chunk ids]}.	None
indexed_ids		the chunk ids really stored in your vector index.	None
index_results		{query_id: [chunk ids the index returned, best first]}.	None
k		how many neighbours to compare.	5

Returns:

Type	Description
	A dict of metric name -> value (None = not measured).

Source code in src/rag_titan/evaluation/stages/embedding.py

def evaluate(self, vectors, query_vectors=None, relevant_ids=None, indexed_ids=None,
             index_results=None, k=5):
    """
    Score the vectors.

    Args:
        vectors: {chunk_id: vector} of every chunk.
        query_vectors: {query_id: vector}.
        relevant_ids: {query_id: [gold chunk ids]}.
        indexed_ids: the chunk ids really stored in your vector index.
        index_results: {query_id: [chunk ids the index returned, best first]}.
        k: how many neighbours to compare.

    Returns:
        A dict of metric name -> value (None = not measured).
    """
    check_k(k)
    if not isinstance(vectors, dict):
        raise ValueError("vectors must be a dict of {chunk_id: vector}")

    scores = self._health(vectors)
    scores["index_coverage"] = None
    if indexed_ids is not None and vectors:
        scores["index_coverage"] = len(set(vectors) & set(indexed_ids)) / len(vectors)

    scores.update(self._search_scores(vectors, query_vectors or {}, relevant_ids or {},
                                      index_results, k))
    return scores

EndToEndEvaluator

Scores the final answers, task outcomes and request telemetry.

Source code in src/rag_titan/evaluation/stages/end_to_end.py

class EndToEndEvaluator:
    """Scores the final answers, task outcomes and request telemetry."""

    def evaluate(self, queries):
        """
        Score every query that has an "answer".

        Args:
            queries: list of dicts with "id", "answer" and any of
                "expected_answer", "task_success", "latency_ms", "cost_usd",
                "total_tokens".

        Returns:
            A dict of metric name -> value (None = not measured):

                End-to-End P95 Latency = P95(t_request to answer)
                Task Success Rate      = #successful tasks / #labelled tasks
        """
        check_queries(queries)
        observed = [query for query in queries if "answer" in query]
        with_reference = [query for query in observed if query.get("expected_answer") is not None]
        outcomes = [query["task_success"] for query in observed
                    if query.get("task_success") is not None]
        times = self._numbers(observed, "latency_ms")

        return {
            "query_count": len(observed),
            "answer_exact_match": average([exact_match(query["answer"], query["expected_answer"])
                                           for query in with_reference]),
            "answer_f1": average([token_scores(query["answer"],
                                               query["expected_answer"])["answer_f1"]
                                  for query in with_reference]),
            "task_success_rate": average([float(bool(outcome)) for outcome in outcomes]),
            "p50_latency_ms": percentile(times, 50),
            "p95_latency_ms": percentile(times, 95),
            "mean_cost_usd": average(self._numbers(observed, "cost_usd")),
            "mean_total_tokens": average(self._numbers(observed, "total_tokens")),
        }

    def _numbers(self, queries, key):
        """Return the measured values of one telemetry key, checking they are valid."""
        values = []
        for query in queries:
            value = query.get(key)
            if value is None:
                continue
            if not is_number(value) or value < 0:
                raise ValueError(f"Query {query['id']!r}: {key!r} must be a number of 0 or more")
            values.append(value)
        return values

evaluate(queries)

Score every query that has an "answer".

Parameters:

Name	Type	Description	Default
queries		list of dicts with "id", "answer" and any of "expected_answer", "task_success", "latency_ms", "cost_usd", "total_tokens".	required

Returns:

Type	Description
	A dict of metric name -> value (None = not measured): End-to-End P95 Latency = P95(t_request to answer) Task Success Rate = #successful tasks / #labelled tasks

Source code in src/rag_titan/evaluation/stages/end_to_end.py

def evaluate(self, queries):
    """
    Score every query that has an "answer".

    Args:
        queries: list of dicts with "id", "answer" and any of
            "expected_answer", "task_success", "latency_ms", "cost_usd",
            "total_tokens".

    Returns:
        A dict of metric name -> value (None = not measured):

            End-to-End P95 Latency = P95(t_request to answer)
            Task Success Rate      = #successful tasks / #labelled tasks
    """
    check_queries(queries)
    observed = [query for query in queries if "answer" in query]
    with_reference = [query for query in observed if query.get("expected_answer") is not None]
    outcomes = [query["task_success"] for query in observed
                if query.get("task_success") is not None]
    times = self._numbers(observed, "latency_ms")

    return {
        "query_count": len(observed),
        "answer_exact_match": average([exact_match(query["answer"], query["expected_answer"])
                                       for query in with_reference]),
        "answer_f1": average([token_scores(query["answer"],
                                           query["expected_answer"])["answer_f1"]
                              for query in with_reference]),
        "task_success_rate": average([float(bool(outcome)) for outcome in outcomes]),
        "p50_latency_ms": percentile(times, 50),
        "p95_latency_ms": percentile(times, 95),
        "mean_cost_usd": average(self._numbers(observed, "cost_usd")),
        "mean_total_tokens": average(self._numbers(observed, "total_tokens")),
    }

GenerationEvaluator

Scores generated answers against reference answers and your labels.

Source code in src/rag_titan/evaluation/stages/generation.py

class GenerationEvaluator:
    """Scores generated answers against reference answers and your labels."""

    def evaluate(self, queries, chunks=None):
        """
        Score every query that has an "answer".

        Args:
            queries: list of dicts with "id", "answer" and any of:
                "expected_answer", "required_facts" (phrases), "citations" and
                "expected_citations" (source ids), "verified_claims" (list of
                {"supported": True/False}), "answerable" and "abstained",
                "task_success", "generation_latency_ms", and the context as
                "context" (text) or "context_ids".
            chunks: the chunks, needed only to turn "context_ids" into text.

        Returns:
            {"summary": the mean of each metric, "per_query": one dict per query}.
        """
        check_queries(queries)
        items, _ = normalize_chunks(chunks or [], {})
        texts = {item["id"]: item["text"] for item in items}

        per_query = [self._score_query(query, texts) for query in queries if "answer" in query]

        summary = {"query_count": len(per_query)}
        for key in SUMMARY_KEYS:
            summary[key] = average([scores[key] for scores in per_query])
        times = [scores["latency_ms"] for scores in per_query]
        summary["p50_latency_ms"] = percentile(times, 50)
        summary["p95_latency_ms"] = percentile(times, 95)
        return {"summary": summary, "per_query": per_query}

    def _score_query(self, query, texts):
        """Return every generation score of one query."""
        answer = query["answer"]
        if not isinstance(answer, str):
            raise ValueError(f"Query {query['id']!r}: 'answer' must be a text")

        scores = {"query_id": query["id"]}
        scores.update(self._reference_scores(answer, query.get("expected_answer")))
        scores["required_fact_coverage"] = required_fact_coverage(
            answer, query.get("required_facts"))
        scores.update(citation_scores(query.get("citations"), query.get("expected_citations")))
        scores.update(self._label_scores(query))

        context = self._context_text(query, texts)
        scores["context_grounding"] = context_grounding(answer, context)
        scores["unsupported_number_rate"] = unsupported_number_rate(answer, context)
        scores["latency_ms"] = query.get("generation_latency_ms")
        return scores

    def _reference_scores(self, answer, expected):
        """Compare the answer with the reference answer (None when there is none)."""
        if expected is None:
            return {key: None for key in ("exact_match", "answer_precision", "answer_recall",
                                          "answer_f1", "rouge_l_f1")}
        scores = {"exact_match": exact_match(answer, expected)}
        scores.update(token_scores(answer, expected))
        scores["rouge_l_f1"] = rouge_l_f1(answer, expected)
        return scores

    def _label_scores(self, query):
        """
        Scores that come from labels you supply.

            annotated_claim_support  claims marked supported / claims
            Abstention Accuracy      1 when "abstained" is right: the system
                                     refused an unanswerable question or
                                     answered an answerable one
        """
        claims = query.get("verified_claims")
        abstained, answerable = query.get("abstained"), query.get("answerable")

        support, abstention = None, None
        if claims:
            support = sum(bool(claim["supported"]) for claim in claims) / len(claims)
        if abstained is not None and answerable is not None:
            abstention = float(bool(abstained) != bool(answerable))

        success = query.get("task_success")
        return {"annotated_claim_support": support, "abstention_accuracy": abstention,
                "task_success": None if success is None else float(bool(success))}

    def _context_text(self, query, texts):
        """Return the context the LLM was given, or None when it is not known."""
        if query.get("context") is not None:
            return query["context"]

        found = [texts[chunk_id] for chunk_id in query.get("context_ids") or []
                 if chunk_id in texts]
        return " ".join(found) if found else None

evaluate(queries, chunks=None)

Score every query that has an "answer".

Parameters:

Name	Type	Description	Default
queries		list of dicts with "id", "answer" and any of: "expected_answer", "required_facts" (phrases), "citations" and "expected_citations" (source ids), "verified_claims" (list of {"supported": True/False}), "answerable" and "abstained", "task_success", "generation_latency_ms", and the context as "context" (text) or "context_ids".	required
chunks		the chunks, needed only to turn "context_ids" into text.	None

Returns:

Type	Description
	{"summary": the mean of each metric, "per_query": one dict per query}.

Source code in src/rag_titan/evaluation/stages/generation.py

def evaluate(self, queries, chunks=None):
    """
    Score every query that has an "answer".

    Args:
        queries: list of dicts with "id", "answer" and any of:
            "expected_answer", "required_facts" (phrases), "citations" and
            "expected_citations" (source ids), "verified_claims" (list of
            {"supported": True/False}), "answerable" and "abstained",
            "task_success", "generation_latency_ms", and the context as
            "context" (text) or "context_ids".
        chunks: the chunks, needed only to turn "context_ids" into text.

    Returns:
        {"summary": the mean of each metric, "per_query": one dict per query}.
    """
    check_queries(queries)
    items, _ = normalize_chunks(chunks or [], {})
    texts = {item["id"]: item["text"] for item in items}

    per_query = [self._score_query(query, texts) for query in queries if "answer" in query]

    summary = {"query_count": len(per_query)}
    for key in SUMMARY_KEYS:
        summary[key] = average([scores[key] for scores in per_query])
    times = [scores["latency_ms"] for scores in per_query]
    summary["p50_latency_ms"] = percentile(times, 50)
    summary["p95_latency_ms"] = percentile(times, 95)
    return {"summary": summary, "per_query": per_query}

RAGEvaluator

Runs the stage evaluators that have data, adds root-cause hints and a scorecard.

Source code in src/rag_titan/evaluation/rag_evaluator.py

class RAGEvaluator:
    """Runs the stage evaluators that have data, adds root-cause hints and a scorecard."""

    def __init__(self, k=5, thresholds=None, scorer=None):
        """
        Args:
            k: how many results to score in the ranking metrics (Recall@K, ...).
            thresholds: {metric: minimum} for the root-cause hints.
            scorer: a ScoreAggregator with your own weights, targets or stages.
        """
        self.k = check_k(k)
        self.analyzer = RootCauseAnalyzer(thresholds)
        self.scorer = scorer or ScoreAggregator()

    def evaluate(self, chunks=None, queries=None, document_text=None, documents=None,
                 embeddings=None, max_context_tokens=None, **chunk_options):
        """
        Evaluate every stage that has data.

        Args:
            chunks: the chunks of your index (see ChunkEvaluator).
            queries: list of query dicts. Each stage reads its own keys:
                "retrieved_ids", "reranked_ids", "context_ids", "answer", ...
            document_text: the source text of a single document.
            documents: {doc_id: text} when there are several documents.
            embeddings: dict with "vectors" and optionally "query_vectors",
                "relevant_ids", "indexed_ids", "index_results".
            max_context_tokens: the context budget in words.
            **chunk_options: any other ChunkEvaluator input, e.g.
                expected_evidence=[...], size_limits=(50, 300).

        Returns:
            A report dict: one entry per evaluated stage, plus "root_causes"
            and "scorecard".
        """
        unknown = sorted(set(chunk_options) - set(CHUNK_OPTIONS))
        if unknown:
            raise ValueError(f"Unknown input {unknown}: use one of {CHUNK_OPTIONS}")
        queries = check_queries(queries or [])
        source = {"document_text": document_text, "documents": documents}

        report = {}
        if chunks is not None:
            if embeddings and "vectors" in embeddings:
                chunk_options.setdefault("chunk_vectors", embeddings["vectors"])
            report["chunking"] = ChunkEvaluator().evaluate(chunks, **source, **chunk_options)
        if embeddings is not None:
            report["embedding"] = self._embedding(embeddings)

        report.update(self._query_stages(chunks or [], queries, source, max_context_tokens))
        report["root_causes"] = self.analyzer.evaluate(report)
        report["scorecard"] = self.scorer.evaluate(report)
        return report

    def _embedding(self, embeddings):
        """Check the "embeddings" input and run the embedding stage."""
        if not isinstance(embeddings, dict) or "vectors" not in embeddings:
            raise ValueError("embeddings must be a dict with 'vectors'")
        unknown = sorted(set(embeddings) - set(EMBEDDING_KEYS))
        if unknown:
            raise ValueError(f"Unknown embeddings key {unknown}: use one of {EMBEDDING_KEYS}")
        return EmbeddingEvaluator().evaluate(**embeddings, k=self.k)

    def _query_stages(self, chunks, queries, source, max_context_tokens):
        """Run each stage that reads the queries, when a query has its data."""

        def has(key):
            return any(key in query for query in queries)

        report = {}
        if has("retrieved_ids"):
            report["retrieval"] = RetrievalEvaluator().evaluate(chunks, queries, self.k, **source)
        if has("reranked_ids"):
            report["reranking"] = RerankingEvaluator().evaluate(queries, self.k)
        if has("context_ids"):
            report["augmentation"] = AugmentationEvaluator().evaluate(
                chunks, queries, max_context_tokens=max_context_tokens, **source)
        if has("answer"):
            report["generation"] = GenerationEvaluator().evaluate(queries, chunks)
            report["end_to_end"] = EndToEndEvaluator().evaluate(queries)
        return report

__init__(k=5, thresholds=None, scorer=None)

Parameters:

Name	Type	Description	Default
k		how many results to score in the ranking metrics (Recall@K, ...).	5
thresholds		{metric: minimum} for the root-cause hints.	None
scorer		a ScoreAggregator with your own weights, targets or stages.	None

Source code in src/rag_titan/evaluation/rag_evaluator.py

def __init__(self, k=5, thresholds=None, scorer=None):
    """
    Args:
        k: how many results to score in the ranking metrics (Recall@K, ...).
        thresholds: {metric: minimum} for the root-cause hints.
        scorer: a ScoreAggregator with your own weights, targets or stages.
    """
    self.k = check_k(k)
    self.analyzer = RootCauseAnalyzer(thresholds)
    self.scorer = scorer or ScoreAggregator()

evaluate(chunks=None, queries=None, document_text=None, documents=None, embeddings=None, max_context_tokens=None, **chunk_options)

Evaluate every stage that has data.

Parameters:

Name	Type	Description	Default
chunks		the chunks of your index (see ChunkEvaluator).	None
queries		list of query dicts. Each stage reads its own keys: "retrieved_ids", "reranked_ids", "context_ids", "answer", ...	None
document_text		the source text of a single document.	None
documents		{doc_id: text} when there are several documents.	None
embeddings		dict with "vectors" and optionally "query_vectors", "relevant_ids", "indexed_ids", "index_results".	None
max_context_tokens		the context budget in words.	None
**chunk_options		any other ChunkEvaluator input, e.g. expected_evidence=[...], size_limits=(50, 300).	{}

Returns:

Type	Description
	A report dict: one entry per evaluated stage, plus "root_causes"
	and "scorecard".

Source code in src/rag_titan/evaluation/rag_evaluator.py

def evaluate(self, chunks=None, queries=None, document_text=None, documents=None,
             embeddings=None, max_context_tokens=None, **chunk_options):
    """
    Evaluate every stage that has data.

    Args:
        chunks: the chunks of your index (see ChunkEvaluator).
        queries: list of query dicts. Each stage reads its own keys:
            "retrieved_ids", "reranked_ids", "context_ids", "answer", ...
        document_text: the source text of a single document.
        documents: {doc_id: text} when there are several documents.
        embeddings: dict with "vectors" and optionally "query_vectors",
            "relevant_ids", "indexed_ids", "index_results".
        max_context_tokens: the context budget in words.
        **chunk_options: any other ChunkEvaluator input, e.g.
            expected_evidence=[...], size_limits=(50, 300).

    Returns:
        A report dict: one entry per evaluated stage, plus "root_causes"
        and "scorecard".
    """
    unknown = sorted(set(chunk_options) - set(CHUNK_OPTIONS))
    if unknown:
        raise ValueError(f"Unknown input {unknown}: use one of {CHUNK_OPTIONS}")
    queries = check_queries(queries or [])
    source = {"document_text": document_text, "documents": documents}

    report = {}
    if chunks is not None:
        if embeddings and "vectors" in embeddings:
            chunk_options.setdefault("chunk_vectors", embeddings["vectors"])
        report["chunking"] = ChunkEvaluator().evaluate(chunks, **source, **chunk_options)
    if embeddings is not None:
        report["embedding"] = self._embedding(embeddings)

    report.update(self._query_stages(chunks or [], queries, source, max_context_tokens))
    report["root_causes"] = self.analyzer.evaluate(report)
    report["scorecard"] = self.scorer.evaluate(report)
    return report

ReportManager

One place for everything you do with a finished report.

Source code in src/rag_titan/evaluation/reports/manager.py

class ReportManager:
    """One place for everything you do with a finished report."""

    def save_json(self, report, path):
        """Write the report as JSON (folders are created) and return the path."""
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8")
        return path

    def load_json(self, path):
        """Read a report saved with save_json()."""
        path = Path(path)
        if not path.is_file():
            raise FileNotFoundError(f"Report file not found: {path}")
        try:
            report = json.loads(path.read_text(encoding="utf-8"))
        except json.JSONDecodeError as error:
            raise ValueError(f"{path} is not valid JSON (line {error.lineno}): {error.msg}")
        if not isinstance(report, dict):
            raise ValueError(f"{path} does not hold a report (a JSON object)")
        return report

    def markdown(self, report, title="RAG Pipeline Evaluation", generated=None):
        """Return the report as Markdown text (see to_markdown)."""
        return to_markdown(report, title, generated)

    def save_markdown(self, report, path, title="RAG Pipeline Evaluation"):
        """Write the report as Markdown (folders are created) and return the path."""
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(self.markdown(report, title), encoding="utf-8")
        return path

    def flatten(self, report):
        """Return every numeric stage metric as {"stage.metric": value}."""
        return flatten(report)

    def regressions(self, report, baseline, tolerance=0.02, latency_share=LATENCY_SHARE,
                    latency_min_ms=LATENCY_MIN_MS):
        """List the metrics that got worse than in the baseline (see find_regressions)."""
        return find_regressions(report, baseline, tolerance, latency_share, latency_min_ms)

    def trends(self, reports):
        """Return the first-to-last change of every metric over reports, oldest first."""
        return trends(reports)

    def failed_queries(self, report):
        """Return the queries that failed a check, earliest stage first."""
        return failed_queries(report)

failed_queries(report)

Return the queries that failed a check, earliest stage first.

Source code in src/rag_titan/evaluation/reports/manager.py

def failed_queries(self, report):
    """Return the queries that failed a check, earliest stage first."""
    return failed_queries(report)

flatten(report)

Return every numeric stage metric as {"stage.metric": value}.

Source code in src/rag_titan/evaluation/reports/manager.py

def flatten(self, report):
    """Return every numeric stage metric as {"stage.metric": value}."""
    return flatten(report)

load_json(path)

Read a report saved with save_json().

Source code in src/rag_titan/evaluation/reports/manager.py

def load_json(self, path):
    """Read a report saved with save_json()."""
    path = Path(path)
    if not path.is_file():
        raise FileNotFoundError(f"Report file not found: {path}")
    try:
        report = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as error:
        raise ValueError(f"{path} is not valid JSON (line {error.lineno}): {error.msg}")
    if not isinstance(report, dict):
        raise ValueError(f"{path} does not hold a report (a JSON object)")
    return report

markdown(report, title='RAG Pipeline Evaluation', generated=None)

Return the report as Markdown text (see to_markdown).

Source code in src/rag_titan/evaluation/reports/manager.py

def markdown(self, report, title="RAG Pipeline Evaluation", generated=None):
    """Return the report as Markdown text (see to_markdown)."""
    return to_markdown(report, title, generated)

regressions(report, baseline, tolerance=0.02, latency_share=LATENCY_SHARE, latency_min_ms=LATENCY_MIN_MS)

List the metrics that got worse than in the baseline (see find_regressions).

Source code in src/rag_titan/evaluation/reports/manager.py

def regressions(self, report, baseline, tolerance=0.02, latency_share=LATENCY_SHARE,
                latency_min_ms=LATENCY_MIN_MS):
    """List the metrics that got worse than in the baseline (see find_regressions)."""
    return find_regressions(report, baseline, tolerance, latency_share, latency_min_ms)

save_json(report, path)

Write the report as JSON (folders are created) and return the path.

Source code in src/rag_titan/evaluation/reports/manager.py

def save_json(self, report, path):
    """Write the report as JSON (folders are created) and return the path."""
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8")
    return path

save_markdown(report, path, title='RAG Pipeline Evaluation')

Write the report as Markdown (folders are created) and return the path.

Source code in src/rag_titan/evaluation/reports/manager.py

def save_markdown(self, report, path, title="RAG Pipeline Evaluation"):
    """Write the report as Markdown (folders are created) and return the path."""
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(self.markdown(report, title), encoding="utf-8")
    return path

trends(reports)

Return the first-to-last change of every metric over reports, oldest first.

Source code in src/rag_titan/evaluation/reports/manager.py

def trends(self, reports):
    """Return the first-to-last change of every metric over reports, oldest first."""
    return trends(reports)

RerankingEvaluator

Compares the order before ("retrieved_ids") and after ("reranked_ids") reranking.

Source code in src/rag_titan/evaluation/stages/reranking.py

class RerankingEvaluator:
    """Compares the order before ("retrieved_ids") and after ("reranked_ids") reranking."""

    def evaluate(self, queries, k=5):
        """
        Score every query that has "reranked_ids".

        Args:
            queries: list of dicts with "id", "reranked_ids" (best first), the
                gold labels ("expected_chunk_ids" or graded "relevance") and,
                to measure the gains, "retrieved_ids".
            k: how many results to score per query.

        Returns:
            {"summary": the mean of each metric, "per_query": one dict per query}.
            Gains are None for a query without "retrieved_ids".
        """
        check_k(k)
        check_queries(queries)
        per_query = [self._score_query(query, k) for query in queries if "reranked_ids" in query]

        summary = {"query_count": len(per_query)}
        for name, key in SUMMARY_KEYS.items():
            summary[name] = average([scores[key] for scores in per_query])
        return {"summary": summary, "per_query": per_query}

    def _score_query(self, query, k):
        """Score one query before and after reranking."""
        relevance = relevance_of(query)
        reranked = ranked_ids(query, "reranked_ids")
        after = ranking_scores(reranked, relevance, k)

        retrieved = ranked_ids(query, "retrieved_ids") if "retrieved_ids" in query else None
        before = {key: None for key in RANKING_KEYS}
        if retrieved is not None:
            before = ranking_scores(retrieved, relevance, k)

        scores = {"query_id": query["id"]}
        for name in ("ndcg", "mrr", "precision"):
            key = f"{name}_at_k"
            scores[f"before_{name}"] = before[key]
            scores[f"after_{name}"] = after[key]
            scores[f"{name}_gain"] = difference(after[key], before[key])

        scores["first_relevant_rank_change"] = self._rank_change(retrieved, reranked, relevance)
        scores["candidate_set_preserved"] = self._preserved(retrieved, reranked)
        return scores

    def _rank_change(self, retrieved, reranked, relevance):
        """
        How many places the first relevant chunk moved up (whole lists, not only top k).
        +2 means it went from rank 3 to rank 1; a negative number means it moved down.
        """
        if retrieved is None:
            return None
        before = first_relevant_rank(retrieved, relevance)
        after = first_relevant_rank(reranked, relevance)
        return difference(before, after)

    def _preserved(self, retrieved, reranked):
        """
        1.0 when the reranker returned exactly the chunks it was given (only reordered).
        A reranker that also cuts the list gives 0.0, which can be what you want.
        """
        if retrieved is None:
            return None
        return float(set(retrieved) == set(reranked))

evaluate(queries, k=5)

Score every query that has "reranked_ids".

Parameters:

Name	Type	Description	Default
queries		list of dicts with "id", "reranked_ids" (best first), the gold labels ("expected_chunk_ids" or graded "relevance") and, to measure the gains, "retrieved_ids".	required
k		how many results to score per query.	5

Returns:

Type	Description
	{"summary": the mean of each metric, "per_query": one dict per query}.
	Gains are None for a query without "retrieved_ids".

Source code in src/rag_titan/evaluation/stages/reranking.py

def evaluate(self, queries, k=5):
    """
    Score every query that has "reranked_ids".

    Args:
        queries: list of dicts with "id", "reranked_ids" (best first), the
            gold labels ("expected_chunk_ids" or graded "relevance") and,
            to measure the gains, "retrieved_ids".
        k: how many results to score per query.

    Returns:
        {"summary": the mean of each metric, "per_query": one dict per query}.
        Gains are None for a query without "retrieved_ids".
    """
    check_k(k)
    check_queries(queries)
    per_query = [self._score_query(query, k) for query in queries if "reranked_ids" in query]

    summary = {"query_count": len(per_query)}
    for name, key in SUMMARY_KEYS.items():
        summary[name] = average([scores[key] for scores in per_query])
    return {"summary": summary, "per_query": per_query}

RetrievalEvaluator

Scores the ranked chunk ids a retriever returned for each query.

Source code in src/rag_titan/evaluation/stages/retrieval.py

class RetrievalEvaluator:
    """Scores the ranked chunk ids a retriever returned for each query."""

    def evaluate(self, chunks, queries, k=5, document_text=None, documents=None):
        """
        Score every query that has "retrieved_ids".

        Args:
            chunks: all chunks of the index: texts, or dicts with "id", "text"
                and, when you have them, "start", "end" and "doc_id".
            queries: list of dicts with "id", "retrieved_ids" (best first) and
                the gold labels: "expected_chunk_ids" or graded "relevance",
                and optionally "expected_evidence" and "retrieval_latency_ms".
            k: how many results to score per query.
            document_text: the source text of a single document.
            documents: {doc_id: text} when there are several documents.

        Returns:
            {"summary": the mean of each metric, "per_query": one dict per
            query, "warnings": chunks that could not be placed in the text}.
            A metric is None when its inputs were not given.
        """
        check_k(k)
        check_queries(queries)
        documents = as_documents(document_text, documents)
        items, warnings = normalize_chunks(chunks, documents)

        by_id = {item["id"]: item for item in items}
        tokens = document_tokens(documents)
        per_query = [self._score_query(query, by_id, documents, tokens, k)
                     for query in queries if "retrieved_ids" in query]

        return {"summary": self._summary(per_query), "per_query": per_query,
                "warnings": warnings}

    def _score_query(self, query, by_id, documents, tokens, k):
        """Return the ranking and token scores of one query."""
        selected = ranked_ids(query, "retrieved_ids")[:k]
        scores = {"query_id": query["id"]}
        scores.update(ranking_scores(selected, relevance_of(query), k))
        scores.update(self._token_scores(query, selected, by_id, documents, tokens))
        scores["latency_ms"] = query.get("retrieval_latency_ms")
        scores["missing_chunk_ids"] = [cid for cid in selected if cid not in by_id]
        return scores

    def _token_scores(self, query, selected, by_id, documents, tokens):
        """
        Compare the words of the retrieved chunks (R) with the gold evidence words (G).

            Token Recall    = |R ∩ G| / |G|
            Token Precision = |R ∩ G| / |R|
            Token IoU       = |R ∩ G| / |R ∪ G|
            Complete Evidence Recall = #gold spans fully inside one chunk / #gold spans
        """
        evidence = evidence_spans(query.get("expected_evidence"), documents) if documents else []
        if not evidence:
            return {key: None for key in TOKEN_KEYS}

        spans = chunk_spans([by_id[cid] for cid in selected if cid in by_id])
        gold = covered_tokens(tokens, evidence)
        found = covered_tokens(tokens, spans)
        both = gold & found

        return {
            "token_recall": len(both) / len(gold) if gold else None,
            "token_precision": len(both) / len(found) if found else None,
            "token_iou": len(both) / len(gold | found) if gold | found else None,
            "complete_evidence_recall":
                sum(is_inside(span, spans) for span in evidence) / len(evidence),
        }

    def _summary(self, per_query):
        """Average every metric over the queries and add the latency percentiles."""
        summary = {"query_count": len(per_query)}
        for key in RANKING_KEYS + TOKEN_KEYS:
            summary[key] = average([scores[key] for scores in per_query])

        times = [scores["latency_ms"] for scores in per_query]
        summary["p50_latency_ms"] = percentile(times, 50)
        summary["p95_latency_ms"] = percentile(times, 95)
        return summary

evaluate(chunks, queries, k=5, document_text=None, documents=None)

Score every query that has "retrieved_ids".

Parameters:

Name	Type	Description	Default
chunks		all chunks of the index: texts, or dicts with "id", "text" and, when you have them, "start", "end" and "doc_id".	required
queries		list of dicts with "id", "retrieved_ids" (best first) and the gold labels: "expected_chunk_ids" or graded "relevance", and optionally "expected_evidence" and "retrieval_latency_ms".	required
k		how many results to score per query.	5
document_text		the source text of a single document.	None
documents		{doc_id: text} when there are several documents.	None

Returns:

Type	Description
	{"summary": the mean of each metric, "per_query": one dict per
	query, "warnings": chunks that could not be placed in the text}.
	A metric is None when its inputs were not given.

Source code in src/rag_titan/evaluation/stages/retrieval.py

def evaluate(self, chunks, queries, k=5, document_text=None, documents=None):
    """
    Score every query that has "retrieved_ids".

    Args:
        chunks: all chunks of the index: texts, or dicts with "id", "text"
            and, when you have them, "start", "end" and "doc_id".
        queries: list of dicts with "id", "retrieved_ids" (best first) and
            the gold labels: "expected_chunk_ids" or graded "relevance",
            and optionally "expected_evidence" and "retrieval_latency_ms".
        k: how many results to score per query.
        document_text: the source text of a single document.
        documents: {doc_id: text} when there are several documents.

    Returns:
        {"summary": the mean of each metric, "per_query": one dict per
        query, "warnings": chunks that could not be placed in the text}.
        A metric is None when its inputs were not given.
    """
    check_k(k)
    check_queries(queries)
    documents = as_documents(document_text, documents)
    items, warnings = normalize_chunks(chunks, documents)

    by_id = {item["id"]: item for item in items}
    tokens = document_tokens(documents)
    per_query = [self._score_query(query, by_id, documents, tokens, k)
                 for query in queries if "retrieved_ids" in query]

    return {"summary": self._summary(per_query), "per_query": per_query,
            "warnings": warnings}

RootCauseAnalyzer

Finds the metrics below their minimum and the first stage that fails.

Source code in src/rag_titan/evaluation/root_cause.py

class RootCauseAnalyzer:
    """Finds the metrics below their minimum and the first stage that fails."""

    def __init__(self, thresholds=None):
        """
        Args:
            thresholds: {metric: minimum} to change the defaults, e.g.
                {"recall_at_k": 0.9, "answer_f1": 0.6}.
        """
        self.thresholds = {metric: minimum for _, metric, minimum, _ in CHECKS}

        for metric, minimum in (thresholds or {}).items():
            if metric not in self.thresholds:
                raise ValueError(f"Unknown threshold {metric!r}: use one of "
                                 f"{sorted(self.thresholds)}")
            if not is_number(minimum):
                raise ValueError(f"Threshold of {metric!r} must be a number, got {minimum!r}")
            self.thresholds[metric] = minimum

    def evaluate(self, report):
        """
        Check a report from RAGEvaluator (or a dict of {stage: stage result}).

        Returns:
            A dict with "first_failing_stage" (a stage name or None), "failures"
            (a list of {"stage", "metric", "observed", "minimum", "suggestion"})
            and "method" (how the hints were made). Metrics that were not
            measured are skipped, not failed.
        """
        if not isinstance(report, dict):
            raise ValueError("report must be a dict of {stage: stage result}")

        failures = []
        for stage, metric, _, suggestion in CHECKS:
            result = report.get(stage) or {}
            observed = result.get("summary", result).get(metric)
            minimum = self.thresholds[metric]
            if observed is not None and observed < minimum:
                failures.append({"stage": stage, "metric": metric, "observed": observed,
                                 "minimum": minimum, "suggestion": suggestion})

        return {"first_failing_stage": failures[0]["stage"] if failures else None,
                "failures": failures, "method": METHOD}

__init__(thresholds=None)

Parameters:

Name	Type	Description	Default
thresholds		{metric: minimum} to change the defaults, e.g. {"recall_at_k": 0.9, "answer_f1": 0.6}.	None

Source code in src/rag_titan/evaluation/root_cause.py

def __init__(self, thresholds=None):
    """
    Args:
        thresholds: {metric: minimum} to change the defaults, e.g.
            {"recall_at_k": 0.9, "answer_f1": 0.6}.
    """
    self.thresholds = {metric: minimum for _, metric, minimum, _ in CHECKS}

    for metric, minimum in (thresholds or {}).items():
        if metric not in self.thresholds:
            raise ValueError(f"Unknown threshold {metric!r}: use one of "
                             f"{sorted(self.thresholds)}")
        if not is_number(minimum):
            raise ValueError(f"Threshold of {metric!r} must be a number, got {minimum!r}")
        self.thresholds[metric] = minimum

evaluate(report)

Check a report from RAGEvaluator (or a dict of {stage: stage result}).

Returns:

Type	Description
	A dict with "first_failing_stage" (a stage name or None), "failures"
	(a list of {"stage", "metric", "observed", "minimum", "suggestion"})
	and "method" (how the hints were made). Metrics that were not
	measured are skipped, not failed.

Source code in src/rag_titan/evaluation/root_cause.py

def evaluate(self, report):
    """
    Check a report from RAGEvaluator (or a dict of {stage: stage result}).

    Returns:
        A dict with "first_failing_stage" (a stage name or None), "failures"
        (a list of {"stage", "metric", "observed", "minimum", "suggestion"})
        and "method" (how the hints were made). Metrics that were not
        measured are skipped, not failed.
    """
    if not isinstance(report, dict):
        raise ValueError("report must be a dict of {stage: stage result}")

    failures = []
    for stage, metric, _, suggestion in CHECKS:
        result = report.get(stage) or {}
        observed = result.get("summary", result).get(metric)
        minimum = self.thresholds[metric]
        if observed is not None and observed < minimum:
            failures.append({"stage": stage, "metric": metric, "observed": observed,
                             "minimum": minimum, "suggestion": suggestion})

    return {"first_failing_stage": failures[0]["stage"] if failures else None,
            "failures": failures, "method": METHOD}

ScoreAggregator

Combines the stage metrics into stage scores, an overall score and a status.

Source code in src/rag_titan/evaluation/scoring/scorecard.py

class ScoreAggregator:
    """Combines the stage metrics into stage scores, an overall score and a status."""

    def __init__(self, stage_weights=None, metric_weights=None, targets=None,
                 quality_gates=None, stages=None, min_coverage=0.75):
        """
        Args:
            stage_weights: {stage: weight} to change how much a stage counts.
            metric_weights: {stage: {metric: weight}}; weight 0 switches a metric off.
            targets: {stage: {metric: target}} for sizes, latency and cost, e.g.
                {"end_to_end": {"p95_latency_ms": 2000}}.
            quality_gates: {stage: {metric: minimum}} replacing the default gates.
            stages: the stages to score, e.g. ["retrieval"]; default is all.
            min_coverage: below this measured share the score is "Provisional".
        """
        self.stages = list(STAGE_WEIGHTS) if stages is None else list(stages)
        unknown = set(self.stages) - set(STAGE_WEIGHTS)
        if unknown or not self.stages or len(self.stages) != len(set(self.stages)):
            raise ValueError(f"stages must be unique names from {list(STAGE_WEIGHTS)}, "
                             f"got {self.stages}")
        if not is_number(min_coverage) or not 0 <= min_coverage <= 1:
            raise ValueError("min_coverage must be between 0 and 1")

        self.stage_weights = {stage: STAGE_WEIGHTS[stage] for stage in self.stages}
        self.stage_weights.update(stage_weights or {})
        self.metric_weights = metric_weights or {}
        self.targets = targets or {}
        self.quality_gates = QUALITY_GATES if quality_gates is None else quality_gates
        self.min_coverage = min_coverage
        self._check_weights()

    def _check_weights(self):
        """Raise a clear error for a weight that is unknown or not a valid number."""
        for stage, weight in self.stage_weights.items():
            if stage not in self.stages or not is_number(weight) or weight <= 0:
                raise ValueError(f"Stage weight of {stage!r} must be a number above 0 "
                                 "for a scored stage")

        for stage, weights in self.metric_weights.items():
            known = set(METRICS.get(stage, {})) | set(TARGET_METRICS.get(stage, {}))
            for name, weight in weights.items():
                if name not in known or not is_number(weight) or weight < 0:
                    raise ValueError(f"Invalid metric weight for {stage}.{name}")

    def evaluate(self, report):
        """
        Score a report from RAGEvaluator (or a dict of {stage: stage result}).

        Returns:
            {"overall": score, rating, status, coverage, gate failures and the
            stages to review, "stages": one entry per stage, "stage_weights", "method"}.
        """
        cards, weighted, measured = {}, 0.0, 0.0
        for stage in self.stages:
            rules = stage_rules(stage, self.metric_weights, self.targets)
            card, score, coverage = score_stage(stage, report.get(stage) or {}, rules,
                                                self.targets.get(stage, {}), self.min_coverage)
            cards[stage] = card
            measured += self.stage_weights[stage] * coverage
            weighted += self.stage_weights[stage] * coverage * (score or 0.0)

        score = weighted / measured if measured else None
        coverage = measured / sum(self.stage_weights.values())
        return {"overall": self._overall(score, coverage, cards, self._gates(report)),
                "stages": cards, "stage_weights": dict(self.stage_weights), "method": METHOD}

    def _gates(self, report):
        """Return the quality gates that failed: measured metrics below their minimum."""
        failures = []
        for stage, gates in self.quality_gates.items():
            if stage not in self.stages:
                continue
            result = report.get(stage) or {}
            values = result.get("summary", result)
            for metric, minimum in gates.items():
                value = values.get(metric)
                if is_number(value) and value < minimum:
                    failures.append({
                        "stage": stage, "metric": metric, "observed": value, "minimum": minimum,
                        "message": f"{STAGE_NAMES[stage][0]}: {metric.replace('_', ' ')} "
                                   "is below the recommended minimum."})
        return failures

    def _overall(self, score, coverage, cards, gates):
        """Build the "overall" entry: score, rating, status and what to review first."""
        label, message = rating(None if score is None else score * 100)
        provisional = coverage < self.min_coverage

        if score is None:
            status = "Not Evaluated"
        elif gates:
            status = "Action Required"
            message += " One or more critical quality checks failed."
        elif provisional:
            status = "Provisional"
        else:
            status = "Evaluated"
            message += " Review individual stages before deciding whether to deploy."
        if provisional and score is not None:
            message += " This is a provisional score because evaluation data is incomplete."

        scored = [stage for stage in self.stages if cards[stage]["score"] is not None]
        weakest = sorted(scored, key=lambda stage: cards[stage]["score"])[:3]
        return {
            "score": None if score is None else round(score * 100, 2),
            "score_0_1": None if score is None else round(score, 4),
            "rating": label, "status": status, "message": message,
            "evaluation_coverage": round(coverage, 4), "is_provisional": provisional,
            "evaluated_stages": len(scored), "total_stages": len(self.stages),
            "quality_gate_failures": gates,
            "areas_to_review": [{"stage": stage, "label": cards[stage]["name"],
                                 "score": cards[stage]["score"],
                                 "improvement_areas": cards[stage]["improvement_areas"]}
                                for stage in weakest],
        }

__init__(stage_weights=None, metric_weights=None, targets=None, quality_gates=None, stages=None, min_coverage=0.75)

Parameters:

Name	Type	Description	Default
stage_weights		{stage: weight} to change how much a stage counts.	None
metric_weights		{stage: {metric: weight}}; weight 0 switches a metric off.	None
targets		{stage: {metric: target}} for sizes, latency and cost, e.g. {"end_to_end": {"p95_latency_ms": 2000}}.	None
quality_gates		{stage: {metric: minimum}} replacing the default gates.	None
stages		the stages to score, e.g. ["retrieval"]; default is all.	None
min_coverage		below this measured share the score is "Provisional".	0.75

Source code in src/rag_titan/evaluation/scoring/scorecard.py

def __init__(self, stage_weights=None, metric_weights=None, targets=None,
             quality_gates=None, stages=None, min_coverage=0.75):
    """
    Args:
        stage_weights: {stage: weight} to change how much a stage counts.
        metric_weights: {stage: {metric: weight}}; weight 0 switches a metric off.
        targets: {stage: {metric: target}} for sizes, latency and cost, e.g.
            {"end_to_end": {"p95_latency_ms": 2000}}.
        quality_gates: {stage: {metric: minimum}} replacing the default gates.
        stages: the stages to score, e.g. ["retrieval"]; default is all.
        min_coverage: below this measured share the score is "Provisional".
    """
    self.stages = list(STAGE_WEIGHTS) if stages is None else list(stages)
    unknown = set(self.stages) - set(STAGE_WEIGHTS)
    if unknown or not self.stages or len(self.stages) != len(set(self.stages)):
        raise ValueError(f"stages must be unique names from {list(STAGE_WEIGHTS)}, "
                         f"got {self.stages}")
    if not is_number(min_coverage) or not 0 <= min_coverage <= 1:
        raise ValueError("min_coverage must be between 0 and 1")

    self.stage_weights = {stage: STAGE_WEIGHTS[stage] for stage in self.stages}
    self.stage_weights.update(stage_weights or {})
    self.metric_weights = metric_weights or {}
    self.targets = targets or {}
    self.quality_gates = QUALITY_GATES if quality_gates is None else quality_gates
    self.min_coverage = min_coverage
    self._check_weights()

evaluate(report)

Score a report from RAGEvaluator (or a dict of {stage: stage result}).

Returns:

Type	Description
	{"overall": score, rating, status, coverage, gate failures and the
	stages to review, "stages": one entry per stage, "stage_weights", "method"}.

Source code in src/rag_titan/evaluation/scoring/scorecard.py

def evaluate(self, report):
    """
    Score a report from RAGEvaluator (or a dict of {stage: stage result}).

    Returns:
        {"overall": score, rating, status, coverage, gate failures and the
        stages to review, "stages": one entry per stage, "stage_weights", "method"}.
    """
    cards, weighted, measured = {}, 0.0, 0.0
    for stage in self.stages:
        rules = stage_rules(stage, self.metric_weights, self.targets)
        card, score, coverage = score_stage(stage, report.get(stage) or {}, rules,
                                            self.targets.get(stage, {}), self.min_coverage)
        cards[stage] = card
        measured += self.stage_weights[stage] * coverage
        weighted += self.stage_weights[stage] * coverage * (score or 0.0)

    score = weighted / measured if measured else None
    coverage = measured / sum(self.stage_weights.values())
    return {"overall": self._overall(score, coverage, cards, self._gates(report)),
            "stages": cards, "stage_weights": dict(self.stage_weights), "method": METHOD}

as_questions(questions)

Accept a questions file path or an already loaded list, and return the checked list. Lets every helper take either "questions.json" or [{...}, ...].

Source code in src/rag_titan/evaluation/questions.py

def as_questions(questions):
    """
    Accept a questions file path or an already loaded list, and return the checked list.
    Lets every helper take either `"questions.json"` or `[{...}, ...]`.
    """
    if isinstance(questions, list):
        return check_questions(questions)
    return load_questions(questions)

build_dataset(chunks, runs, golden, documents, require_validated=True)

Return {"documents", "chunks", "queries", "expected_evidence"} ready for evaluation.

Parameters:

Name	Type	Description	Default
chunks		your chunks, dicts with "id" and "text" ("doc_id" when there are several documents; the text must occur in that document).	required
runs		one dict per question your RAG answered, with "question" (or the golden "id") and its outputs: "retrieved_ids", "context_ids", "answer", latencies, ... (see RUN_KEYS).	required
golden		golden questions, a JSON file path or a list of dicts.	required
documents		{name: text} of the documents your chunks come from.	required
require_validated		refuse golden questions a person has not reviewed.	True

Source code in src/rag_titan/evaluation/dataset.py

def build_dataset(chunks, runs, golden, documents, require_validated=True):
    """
    Return {"documents", "chunks", "queries", "expected_evidence"} ready for evaluation.

    Args:
        chunks: your chunks, dicts with "id" and "text" ("doc_id" when there
            are several documents; the text must occur in that document).
        runs: one dict per question your RAG answered, with "question" (or the
            golden "id") and its outputs: "retrieved_ids", "context_ids",
            "answer", latencies, ... (see RUN_KEYS).
        golden: golden questions, a JSON file path or a list of dicts.
        documents: {name: text} of the documents your chunks come from.
        require_validated: refuse golden questions a person has not reviewed.
    """
    documents = as_documents(documents=documents)
    placed = place_chunks(chunks, documents)
    golden = _check_reviewed(load_golden(golden), require_validated)
    by_id = {record["id"]: record for record in golden}
    by_question = {record["question"].strip(): record for record in golden}

    if not isinstance(runs, list) or not runs:
        raise ValueError("runs must be a non-empty list of dicts")
    queries = [_query(run, number, by_id, by_question, placed, documents)
               for number, run in enumerate(runs, start=1)]
    # every quote once, so chunking can check that no boundary cuts the evidence
    quotes = list(dict.fromkeys(quote for query in queries
                                for quote in query.get("expected_evidence") or []))
    data = {"documents": documents, "chunks": placed, "queries": queries}
    if quotes:
        data["expected_evidence"] = quotes
    return data

check_golden(questions, documents=None)

Check a golden set before you trust the scores made with it.

Parameters:

Name	Type	Description	Default
questions		path to a golden questions JSON file, or a list of dicts.	required
documents		{doc_id: text}; when given, every quote must be found in them.	None

Returns:

Type	Description
	A dict with "question_count", "validated_count", "validated_share",
	"problems" (a list of {"id", "problem"}) and "ready".
	A person marks a question as checked with "validated": true.
	"ready" is True when there is no problem and every question is validated.

Source code in src/rag_titan/evaluation/golden.py

def check_golden(questions, documents=None):
    """
    Check a golden set before you trust the scores made with it.

    Args:
        questions: path to a golden questions JSON file, or a list of dicts.
        documents: {doc_id: text}; when given, every quote must be found in them.

    Returns:
        A dict with "question_count", "validated_count", "validated_share",
        "problems" (a list of {"id", "problem"}) and "ready".
        A person marks a question as checked with "validated": true.
        "ready" is True when there is no problem and every question is validated.
    """
    questions = load_golden(questions)
    problems = []
    for record in questions:
        problems += [{"id": record["id"], "problem": problem}
                     for problem in _record_problems(record, documents)]

    validated = sum(record.get("validated") is True for record in questions)
    return {
        "question_count": len(questions),
        "validated_count": validated,
        "validated_share": validated / len(questions),
        "problems": problems,
        "ready": not problems and validated == len(questions),
    }

check_questions(records)

Check a list of question dicts and return it. Raises ValueError with the record id when something is missing or wrong.

Source code in src/rag_titan/evaluation/questions.py

def check_questions(records):
    """
    Check a list of question dicts and return it.
    Raises ValueError with the record id when something is missing or wrong.
    """
    if not isinstance(records, list) or not records:
        raise ValueError("Questions must be a non-empty list of records")

    seen_ids = set()
    for number, record in enumerate(records, start=1):
        _check_record(record, number)
        if record["id"] in seen_ids:
            raise ValueError(f"Question id {record['id']!r} is used more than once")
        seen_ids.add(record["id"])

    return records

collect_dataset(rag, files, questions, include_embeddings=False)

Index the files, run every golden question, and return the RAGEvaluator input.

Parameters:

Name	Type	Description	Default
rag		a RAGTitan.	required
files		list of file paths to index (the index is rebuilt).	required
questions		path to a golden questions JSON file, or a list of dicts.	required
include_embeddings		also collect the chunk and query vectors for the embedding stage (embeds every question once more).	False

Returns:

Type	Description
	A dict with "documents", "chunks", "queries" and, when asked,
	"embeddings": pass it to RAGEvaluator().evaluate(**dataset).

Source code in src/rag_titan/evaluation/collect.py

def collect_dataset(rag, files, questions, include_embeddings=False):
    """
    Index the files, run every golden question, and return the RAGEvaluator input.

    Args:
        rag: a RAGTitan.
        files: list of file paths to index (the index is rebuilt).
        questions: path to a golden questions JSON file, or a list of dicts.
        include_embeddings: also collect the chunk and query vectors for the
            embedding stage (embeds every question once more).

    Returns:
        A dict with "documents", "chunks", "queries" and, when asked,
        "embeddings": pass it to RAGEvaluator().evaluate(**dataset).
    """
    questions = load_golden(questions)
    if not isinstance(files, list) or not files:
        raise ValueError("files must be a non-empty list of file paths")

    loaded = rag.pipeline.load_documents(files)
    documents = {document["source"]: document["text"] for document in loaded}
    if len(documents) != len(loaded):
        raise ValueError("Two files have the same name: rename one of them")
    rag.index(files)

    raw_chunks = rag.pipeline.chunker.chunk(loaded)
    chunks = [_as_eval_chunk(chunk, documents) for chunk in raw_chunks]
    queries = [_run_question(rag, record, gold_chunk_ids(record, raw_chunks, documents))
               for record in questions]

    dataset = {"documents": documents, "chunks": chunks, "queries": queries}
    if include_embeddings:
        dataset["embeddings"] = _embeddings(rag, queries)
    return dataset

compare_retrievers(retrievers, questions, top_k=5)

Evaluate several retrievers on identical questions and top_k.

Parameters:

Name	Type	Description	Default
retrievers		dict of {name: retriever}, e.g. {"dense": r1, "bm25": r2}.	required

Returns:

Type	Description
	A list with one evaluate_retriever() result per retriever, in order.

Source code in src/rag_titan/evaluation/benchmark.py

def compare_retrievers(retrievers, questions, top_k=5):
    """
    Evaluate several retrievers on identical questions and top_k.

    Args:
        retrievers: dict of {name: retriever}, e.g. {"dense": r1, "bm25": r2}.

    Returns:
        A list with one evaluate_retriever() result per retriever, in order.
    """
    if not isinstance(retrievers, dict) or not retrievers:
        raise ValueError("retrievers must be a non-empty dict of {name: retriever}")

    return [evaluate_retriever(retriever, questions, top_k, name=name)
            for name, retriever in retrievers.items()]

compare_setups(files, questions, setups)

Index the files and evaluate every setup on the same questions.

Parameters:

Name	Type	Description	Default
files		list of file paths to index.	required
questions		path to a questions file, or a list of question dicts.	required
setups		list of {"name": ..., "rag_type": "simple", "settings": {...}}. "settings" changes any setting (chunker, chunk_size, models, ...).	required

Returns:

Type	Description
	A list with one evaluate_rag() result per setup, in order; each also
	has the setup's "rag_type" and "settings".

Setups with the same settings share one index, so comparing RAG types on the same ingestion embeds the files only once.

Source code in src/rag_titan/evaluation/end_to_end.py

def compare_setups(files, questions, setups):
    """
    Index the files and evaluate every setup on the same questions.

    Args:
        files: list of file paths to index.
        questions: path to a questions file, or a list of question dicts.
        setups: list of {"name": ..., "rag_type": "simple", "settings": {...}}.
            "settings" changes any setting (chunker, chunk_size, models, ...).

    Returns:
        A list with one evaluate_rag() result per setup, in order; each also
        has the setup's "rag_type" and "settings".

    Setups with the same settings share one index, so comparing RAG types
    on the same ingestion embeds the files only once.
    """
    if not isinstance(files, list) or not files:
        raise ValueError("files must be a non-empty list of file paths")
    questions = as_questions(questions)
    _check_setups(setups)

    indexes = {}
    results = []
    for setup in setups:
        parts = _indexed_parts(files, setup.get("settings") or {}, indexes)
        rag = build_rag(setup["rag_type"], parts)

        result = evaluate_rag(rag, questions, name=setup["name"])
        result["rag_type"] = setup["rag_type"]
        result["settings"] = setup.get("settings") or {}
        results.append(result)

    return results

evaluate_file(input_path, output_path='rag_report.json', k=5, baseline=None)

Evaluate a data file and save the report as JSON and Markdown.

Parameters:

Name	Type	Description	Default
input_path		the JSON data file.	required
output_path		where to write the JSON report; the Markdown report gets the same name ending in .md.	'rag_report.json'
k		how many results to score in the ranking metrics.	5
baseline		path of an earlier JSON report; the metrics that got worse are added to the report as "regressions".	None

Returns:

Type	Description
	The report dict.

Source code in src/rag_titan/evaluation/reports/run.py

def evaluate_file(input_path, output_path="rag_report.json", k=5, baseline=None):
    """
    Evaluate a data file and save the report as JSON and Markdown.

    Args:
        input_path: the JSON data file.
        output_path: where to write the JSON report; the Markdown report gets
            the same name ending in .md.
        k: how many results to score in the ranking metrics.
        baseline: path of an earlier JSON report; the metrics that got worse
            are added to the report as "regressions".

    Returns:
        The report dict.
    """
    manager = ReportManager()
    report = RAGEvaluator(k=k).evaluate(**load_dataset(input_path))
    if baseline:
        report["regressions"] = manager.regressions(report, manager.load_json(baseline))

    manager.save_json(report, output_path)
    manager.save_markdown(report, Path(output_path).with_suffix(".md"))
    return report

evaluate_rag(rag, questions, name=None)

Answer every question with rag.run() and score the run.

Parameters:

Name	Type	Description	Default
rag		any BaseRAG (SimpleRAG, AdaptiveRAG, ...) with indexed documents.	required
questions		path to a questions file, or a list of question dicts.	required
name		name shown in the report, defaults to the class name.	None

Returns:

Type	Description
	A dict with the means ("recall", "mrr", "ndcg" on the sources,
	"answered" share, average "steps"), the total time per question
	("p50_s", "p95_s") and one row per question in "rows"
	(each row keeps the answer and every step).

Source code in src/rag_titan/evaluation/end_to_end.py

def evaluate_rag(rag, questions, name=None):
    """
    Answer every question with rag.run() and score the run.

    Args:
        rag: any BaseRAG (SimpleRAG, AdaptiveRAG, ...) with indexed documents.
        questions: path to a questions file, or a list of question dicts.
        name: name shown in the report, defaults to the class name.

    Returns:
        A dict with the means ("recall", "mrr", "ndcg" on the sources,
        "answered" share, average "steps"), the total time per question
        ("p50_s", "p95_s") and one row per question in "rows"
        (each row keeps the answer and every step).
    """
    questions = as_questions(questions)
    rows = [_run_question(rag, question) for question in questions]

    summary = summarize_rows(rows)
    result = {"name": name or type(rag).__name__}
    result.update({key: summary[key] for key in ("recall", "mrr", "ndcg")})
    result["answered"] = sum(row["answered"] for row in rows) / len(rows)
    result["steps"] = sum(len(row["steps"]) for row in rows) / len(rows)
    result["p50_s"] = summary["p50_ms"] / 1000
    result["p95_s"] = summary["p95_ms"] / 1000
    result["rows"] = rows
    return result

evaluate_retriever(retriever, questions, top_k=5, name=None)

Ask every question and score what the retriever finds.

Parameters:

Name	Type	Description	Default
retriever		any BaseRetriever with indexed documents.	required
questions		list of question dicts (see load_questions).	required
top_k		how many chunks to retrieve and score per question.	5
name		name shown in the report, defaults to retriever.name.	None

Returns:

Type	Description
	A dict with the means ("recall", "mrr", "ndcg"), the latency
	("p50_ms", "p95_ms") and one row per question in "rows".

Source code in src/rag_titan/evaluation/benchmark.py

def evaluate_retriever(retriever, questions, top_k=5, name=None):
    """
    Ask every question and score what the retriever finds.

    Args:
        retriever: any BaseRetriever with indexed documents.
        questions: list of question dicts (see load_questions).
        top_k: how many chunks to retrieve and score per question.
        name: name shown in the report, defaults to retriever.name.

    Returns:
        A dict with the means ("recall", "mrr", "ndcg"), the latency
        ("p50_ms", "p95_ms") and one row per question in "rows".
    """
    check_questions(questions)
    if not isinstance(top_k, int) or isinstance(top_k, bool) or top_k < 1:
        raise ValueError(f"top_k must be a whole number of at least 1, got {top_k!r}")
    if not retriever.has_documents():
        raise ValueError("The retriever has no documents: index your files first")

    rows = [_score_question(retriever, question, top_k) for question in questions]

    result = {"name": name or retriever.name, "top_k": top_k}
    result.update(summarize_rows(rows))
    result["rows"] = rows
    return result

format_rag_report(results)

Return the end-to-end results as a text table, one line per setup.

Source code in src/rag_titan/evaluation/end_to_end.py

def format_rag_report(results):
    """Return the end-to-end results as a text table, one line per setup."""
    if not results:
        raise ValueError("There are no results to report")

    title = (f"End-to-end report: {len(results[0]['rows'])} questions "
             "(evidence scored on the chunks given to the LLM)")
    return format_table(title, "setup", results, RAG_COLUMNS)

format_report(results)

Return the results as a text table, one line per retriever.

Source code in src/rag_titan/evaluation/benchmark.py

def format_report(results):
    """Return the results as a text table, one line per retriever."""
    if not results:
        raise ValueError("There are no results to report")

    top_k = results[0]["top_k"]
    columns = [(title.replace("@k", f"@{top_k}"), key, fmt) for title, key, fmt in COLUMNS]
    title = f"Retrieval report: {len(results[0]['rows'])} questions, top_k={top_k}"
    return format_table(title, "retriever", results, columns)

load_golden(questions)

Return the golden questions from a JSON file path or a list, after checking them.

Source code in src/rag_titan/evaluation/golden.py

def load_golden(questions):
    """Return the golden questions from a JSON file path or a list, after checking them."""
    if not isinstance(questions, list):
        path = Path(questions)
        if not path.is_file():
            raise FileNotFoundError(f"Golden questions file not found: {path}")
        try:
            questions = json.loads(path.read_text(encoding="utf-8"))
        except json.JSONDecodeError as error:
            raise ValueError(f"{path} is not valid JSON (line {error.lineno}): {error.msg}")

    if not isinstance(questions, list) or not questions:
        raise ValueError("Golden questions must be a non-empty list of records")
    check_queries(questions)
    for record in questions:
        if not isinstance(record.get("question"), str) or not record["question"].strip():
            raise ValueError(f"Question {record['id']!r}: 'question' must be a non-empty text")
    return questions

load_questions(path)

Read and check a questions file.

Parameters:

Name	Type	Description	Default
path		path to the JSON file.	required

Returns:

Type	Description
	The list of question dicts.

Source code in src/rag_titan/evaluation/questions.py

def load_questions(path):
    """
    Read and check a questions file.

    Args:
        path: path to the JSON file.

    Returns:
        The list of question dicts.
    """
    path = Path(path)
    if not path.is_file():
        raise FileNotFoundError(f"Questions file not found: {path}")

    try:
        records = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as error:
        raise ValueError(f"{path} is not valid JSON (line {error.lineno}): {error.msg}")

    return check_questions(records)

make_golden_template(questions, documents=None)

Return one golden record per question, with empty labels to fill in.

Parameters:

Name	Type	Description	Default
questions		a list of question texts.	required
documents		optional {name: text}; when given, each record gets a suggested evidence sentence taken word for word from them.	None

Returns:

Type	Description
	A list of dicts with "id" (q1, q2, ...), "question", "expected_answer",
	"expected_evidence", "required_facts", "answerable", "validated" and "_todo".

Source code in src/rag_titan/evaluation/golden_template.py

def make_golden_template(questions, documents=None):
    """
    Return one golden record per question, with empty labels to fill in.

    Args:
        questions: a list of question texts.
        documents: optional {name: text}; when given, each record gets a
            suggested evidence sentence taken word for word from them.

    Returns:
        A list of dicts with "id" (q1, q2, ...), "question", "expected_answer",
        "expected_evidence", "required_facts", "answerable", "validated" and "_todo".
    """
    if not isinstance(questions, list) or not questions:
        raise ValueError("questions must be a non-empty list of texts")
    if documents is not None and not isinstance(documents, dict):
        raise ValueError("documents must be a dict of {name: text}")

    records = []
    for number, question in enumerate(questions, start=1):
        if not isinstance(question, str) or not question.strip():
            raise ValueError(f"Question {number} is empty")
        suggestion = suggest_evidence(question, documents) if documents else None
        records.append({
            "id": f"q{number}",
            "question": question.strip(),
            "expected_answer": "",
            "expected_evidence": [suggestion] if suggestion else [],
            "required_facts": [],
            "answerable": True,
            "validated": False,
            "_todo": TODO,
        })
    return records

matches_evidence(chunk, evidence_id)

Return True when the chunk is (part of) the evidence.

Source code in src/rag_titan/evaluation/metrics.py

def matches_evidence(chunk, evidence_id):
    """Return True when the chunk is (part of) the evidence."""
    source, page = parse_evidence_id(evidence_id)

    if chunk.get("source") != source:
        return False
    if page is None:
        return True
    return page in (chunk.get("page_numbers") or [])

ndcg_at_k(retrieved, required, k)

Normalised discounted cumulative gain (0 to 1): finding evidence early counts more.

DCG  = Σ gain_rank / log2(rank + 1),  gain = new evidences found at that rank
IDCG = the DCG of one new evidence at each of the first min(|Relevant|, k) ranks

One chunk can hold two evidences (say pages 4 and 5), which can beat the ideal, so the result is capped at 1.

Source code in src/rag_titan/evaluation/metrics.py

def ndcg_at_k(retrieved, required, k):
    """
    Normalised discounted cumulative gain (0 to 1): finding evidence early counts more.

        DCG  = Σ gain_rank / log2(rank + 1),  gain = new evidences found at that rank
        IDCG = the DCG of one new evidence at each of the first min(|Relevant|, k) ranks

    One chunk can hold two evidences (say pages 4 and 5), which can beat the
    ideal, so the result is capped at 1.
    """
    hits = evidence_hits(retrieved, required, k)
    dcg = sum(hit / math.log2(rank + 1) for rank, hit in enumerate(hits, start=1))

    ideal_ranks = min(len(set(required)), k)
    idcg = sum(1 / math.log2(rank + 1) for rank in range(1, ideal_ranks + 1))

    return min(1.0, dcg / idcg)

parse_evidence_id(evidence_id)

Split an evidence id into (source, page). page is None for a whole file.

Parameters:

Name	Type	Description	Default
evidence_id		"hr.pdf" or "hr.pdf#p4".	required

Source code in src/rag_titan/evaluation/metrics.py

def parse_evidence_id(evidence_id):
    """
    Split an evidence id into (source, page). page is None for a whole file.

    Args:
        evidence_id: "hr.pdf" or "hr.pdf#p4".
    """
    if not isinstance(evidence_id, str) or not evidence_id.strip():
        raise ValueError(f"An evidence id must be a non-empty text, got {evidence_id!r}")

    source, _, page = evidence_id.strip().partition("#")
    if not page:
        return source, None

    if not page.startswith("p") or not page[1:].isdigit() or int(page[1:]) < 1:
        raise ValueError(f"Evidence id {evidence_id!r}: write the page as '#p4'")
    return source, int(page[1:])

profile_documents(documents, min_words=20, near_duplicate=0.9)

Profile a document collection: sizes, empty and short documents, duplicates, categories.

Parameters:

Name	Type	Description	Default
documents		{doc_id: text} or a list of document dicts.	required
min_words		documents with fewer words are reported as "short".	20
near_duplicate		two documents sharing at least this share of their words (Jaccard) are near duplicates. None skips the check, which compares every pair and is slow above a few thousand documents.	0.9

Returns:

Type	Description
	A dict with the counts and word statistics, "categories" (count per
	category), "exact_duplicates" (groups of ids), "near_duplicates"
	(pairs with their similarity), "issues" ({doc_id: [reasons]}) and
	"clean_share" (documents without any issue / all documents).

Source code in src/rag_titan/evaluation/profiling.py

def profile_documents(documents, min_words=20, near_duplicate=0.9):
    """
    Profile a document collection: sizes, empty and short documents, duplicates, categories.

    Args:
        documents: {doc_id: text} or a list of document dicts.
        min_words: documents with fewer words are reported as "short".
        near_duplicate: two documents sharing at least this share of their
            words (Jaccard) are near duplicates. None skips the check, which
            compares every pair and is slow above a few thousand documents.

    Returns:
        A dict with the counts and word statistics, "categories" (count per
        category), "exact_duplicates" (groups of ids), "near_duplicates"
        (pairs with their similarity), "issues" ({doc_id: [reasons]}) and
        "clean_share" (documents without any issue / all documents).
    """
    if near_duplicate is not None and (not is_number(near_duplicate)
                                       or not 0 < near_duplicate <= 1):
        raise ValueError("near_duplicate must be above 0 and at most 1, or None")

    records = as_records(documents)
    words = {record["id"]: tokenize(record["text"]) for record in records}
    lengths = [len(tokens) for tokens in words.values()]
    exact = _exact_duplicates(words)
    near = _near_duplicates(words, near_duplicate, exact) if near_duplicate else []

    issues = _issues(words, min_words, exact, near)
    return {
        "document_count": len(records),
        "total_words": sum(lengths),
        "mean_words": average(lengths),
        "min_words": min(lengths),
        "max_words": max(lengths),
        "categories": dict(Counter(record["category"] for record in records)),
        "exact_duplicates": exact,
        "near_duplicates": near,
        "issues": issues,
        "clean_share": (len(records) - len(issues)) / len(records),
    }

recall_at_k(retrieved, required, k)

Share of the required evidence found in the first k chunks (0 to 1).

Recall@k = |Relevant ∩ Retrieved_k| / |Relevant|

Source code in src/rag_titan/evaluation/metrics.py

def recall_at_k(retrieved, required, k):
    """
    Share of the required evidence found in the first k chunks (0 to 1).

        Recall@k = |Relevant ∩ Retrieved_k| / |Relevant|
    """
    return sum(evidence_hits(retrieved, required, k)) / len(set(required))

reciprocal_rank(retrieved, required, k)

1 / rank of the first chunk that is evidence, or 0 when none is in the first k. The mean of this over all questions is the MRR.

Source code in src/rag_titan/evaluation/metrics.py

def reciprocal_rank(retrieved, required, k):
    """
    1 / rank of the first chunk that is evidence, or 0 when none is in the first k.
    The mean of this over all questions is the MRR.
    """
    for rank, hit in enumerate(evidence_hits(retrieved, required, k), start=1):
        if hit:
            return 1 / rank
    return 0.0

sample_documents(documents, size, strategy='random', seed=0, profile=None)

Return the ids of a sample of the documents.

Parameters:

Name	Type	Description	Default
documents		{doc_id: text} or a list of document dicts.	required
size		how many documents to pick (at most all of them).	required
strategy		"random", "stratified" or "risk".	'random'
seed		the random seed, for a repeatable sample.	0
profile		a profile_documents() result for "risk"; made when missing.	None

Source code in src/rag_titan/evaluation/sampling.py

def sample_documents(documents, size, strategy="random", seed=0, profile=None):
    """
    Return the ids of a sample of the documents.

    Args:
        documents: {doc_id: text} or a list of document dicts.
        size: how many documents to pick (at most all of them).
        strategy: "random", "stratified" or "risk".
        seed: the random seed, for a repeatable sample.
        profile: a profile_documents() result for "risk"; made when missing.
    """
    records = as_records(documents)
    if not isinstance(size, int) or isinstance(size, bool) or not 1 <= size <= len(records):
        raise ValueError(f"size must be a whole number from 1 to {len(records)}, got {size!r}")
    if strategy not in STRATEGIES:
        raise ValueError(f"Unknown strategy {strategy!r}: use one of {STRATEGIES}")

    rng = random.Random(seed)
    if strategy == "stratified":
        return _stratified(records, size, rng)
    if strategy == "risk":
        return _risk_first(records, size, rng, profile or profile_documents(records))
    return rng.sample([record["id"] for record in records], size)

Stage evaluators: score the logged outputs of each RAG stage against gold labels.

AugmentationEvaluator

Scores the chunks that were packed into the prompt for each query.

Source code in src/rag_titan/evaluation/stages/augmentation.py

class AugmentationEvaluator:
    """Scores the chunks that were packed into the prompt for each query."""

    def evaluate(self, chunks, queries, document_text=None, documents=None,
                 max_context_tokens=None):
        """
        Score every query that has "context_ids".

        Args:
            chunks: all chunks: texts, or dicts with "id", "text", "start", "end", "doc_id".
            queries: list of dicts with "id", "context_ids" and the gold labels:
                "expected_evidence" (needs the source text) and/or
                "expected_chunk_ids". "max_context_tokens" can be set per query.
            document_text: the source text of a single document.
            documents: {doc_id: text} when there are several documents.
            max_context_tokens: the context budget in words, for every query.

        Returns:
            {"summary": the mean of each metric, "per_query": one dict per
            query, "warnings": chunks that could not be placed in the text}.
        """
        check_queries(queries)
        documents = as_documents(document_text, documents)
        items, warnings = normalize_chunks(chunks, documents)
        by_id = {item["id"]: item for item in items}
        tokens = document_tokens(documents)

        per_query = [self._score_query(query, by_id, documents, tokens, max_context_tokens)
                     for query in queries if "context_ids" in query]

        summary = {"query_count": len(per_query)}
        for key in SUMMARY_KEYS:
            summary[key] = average([scores[key] for scores in per_query])
        return {"summary": summary, "per_query": per_query, "warnings": warnings}

    def _score_query(self, query, by_id, documents, tokens, max_context_tokens):
        """Return every context score of one query."""
        selected = ranked_ids(query, "context_ids")
        used = [by_id[chunk_id] for chunk_id in selected if chunk_id in by_id]
        evidence = []
        if documents:
            evidence = evidence_spans(query.get("expected_evidence"), documents)
        budget = query.get("max_context_tokens", max_context_tokens)

        scores = {"query_id": query["id"]}
        scores.update(self._evidence_scores(used, evidence, tokens))
        scores.update(self._budget_scores(used, tokens, documents, budget))
        scores["evidence_edge_placement"] = self._edge_placement(used, evidence, query)
        scores["missing_chunk_ids"] = [cid for cid in selected if cid not in by_id]
        return scores

    def _evidence_scores(self, used, evidence, tokens):
        """
        Compare the context words (C) with the gold evidence words (G).

            Context Precision  = |C ∩ G| / |C|
            Context Recall     = |C ∩ G| / |G|
            Context Noise      = 1 - Context Precision
            Context Sufficient = 1 when every gold span is fully inside a context chunk
        """
        if not evidence:
            return {key: None for key in SUMMARY_KEYS[:4]}

        spans = chunk_spans(used)
        context, gold = covered_tokens(tokens, spans), covered_tokens(tokens, evidence)
        precision = len(context & gold) / len(context) if context else None
        return {
            "context_precision": precision,
            "context_recall": len(context & gold) / len(gold) if gold else None,
            "context_noise": None if precision is None else 1 - precision,
            "context_sufficient": float(all(is_inside(span, spans) for span in evidence)),
        }

    def _budget_scores(self, used, tokens, documents, budget):
        """
        Size of the context in words, repeated words, and the budget check.

            Duplicate Ratio  = (context words - distinct source words) / context words
            Budget Compliant = 1 when context words <= max_context_tokens
        """
        count = sum(len(tokenize(item["text"])) for item in used)
        scores = {"context_tokens": count, "duplicate_ratio": None, "budget_compliant": None}

        placed = documents and all(item["start"] is not None for item in used)
        if count and placed:
            distinct = len(covered_tokens(tokens, chunk_spans(used)))
            scores["duplicate_ratio"] = max(0.0, (count - distinct) / count)
        if budget is not None:
            scores["budget_compliant"] = float(count <= budget)
        return scores

    def _edge_placement(self, used, evidence, query):
        """
        Is the evidence at the start or end of the context (1.0) or in the middle (0.0)?
        LLMs often overlook text in the middle of a long prompt ("lost in the middle").
        """
        gold_ids = {cid for cid, grade in relevance_of(query).items() if grade > 0}
        placements = []
        for number, item in enumerate(used):
            touches = item["start"] is not None and any(
                doc_id == item["doc_id"] and item["start"] < end and item["end"] > start
                for doc_id, start, end in evidence)
            if item["id"] in gold_ids or touches:
                position = number / (len(used) - 1) if len(used) > 1 else 0.0
                placements.append(abs(2 * position - 1))
        return average(placements)

evaluate(chunks, queries, document_text=None, documents=None, max_context_tokens=None)

Score every query that has "context_ids".

Parameters:

Name	Type	Description	Default
chunks		all chunks: texts, or dicts with "id", "text", "start", "end", "doc_id".	required
queries		list of dicts with "id", "context_ids" and the gold labels: "expected_evidence" (needs the source text) and/or "expected_chunk_ids". "max_context_tokens" can be set per query.	required
document_text		the source text of a single document.	None
documents		{doc_id: text} when there are several documents.	None
max_context_tokens		the context budget in words, for every query.	None

Returns:

Type	Description
	{"summary": the mean of each metric, "per_query": one dict per
	query, "warnings": chunks that could not be placed in the text}.

Source code in src/rag_titan/evaluation/stages/augmentation.py

def evaluate(self, chunks, queries, document_text=None, documents=None,
             max_context_tokens=None):
    """
    Score every query that has "context_ids".

    Args:
        chunks: all chunks: texts, or dicts with "id", "text", "start", "end", "doc_id".
        queries: list of dicts with "id", "context_ids" and the gold labels:
            "expected_evidence" (needs the source text) and/or
            "expected_chunk_ids". "max_context_tokens" can be set per query.
        document_text: the source text of a single document.
        documents: {doc_id: text} when there are several documents.
        max_context_tokens: the context budget in words, for every query.

    Returns:
        {"summary": the mean of each metric, "per_query": one dict per
        query, "warnings": chunks that could not be placed in the text}.
    """
    check_queries(queries)
    documents = as_documents(document_text, documents)
    items, warnings = normalize_chunks(chunks, documents)
    by_id = {item["id"]: item for item in items}
    tokens = document_tokens(documents)

    per_query = [self._score_query(query, by_id, documents, tokens, max_context_tokens)
                 for query in queries if "context_ids" in query]

    summary = {"query_count": len(per_query)}
    for key in SUMMARY_KEYS:
        summary[key] = average([scores[key] for scores in per_query])
    return {"summary": summary, "per_query": per_query, "warnings": warnings}

ChunkEvaluator

Scores the output of a chunker against the source text and gold labels.

Source code in src/rag_titan/evaluation/stages/chunking.py

class ChunkEvaluator:
    """Scores the output of a chunker against the source text and gold labels."""

    def evaluate(self, chunks, document_text=None, documents=None, expected_chunks=None,
                 expected_evidence=None, structural_blocks=None, required_metadata=None,
                 expected_metadata=None, size_limits=None, boundary_tolerance=0,
                 sentence_vectors=None, chunk_vectors=None):
        """
        Score the chunks.

        Args:
            chunks: texts, or dicts with "id", "text", "start", "end", "doc_id", "metadata".
            document_text: the source text of a single document.
            documents: {doc_id: text} when there are several documents.
            expected_chunks: the ideal chunks marked by a human (for boundary F1).
            expected_evidence: gold quotes or {"start", "end"} spans that must stay whole.
            structural_blocks: tables, lists or code that must stay whole (same format).
            required_metadata: keys every chunk must have, e.g. ["source", "page"].
            expected_metadata: {chunk_id: {key: value}} of correct metadata values.
            size_limits: (min, max) words of a valid chunk.
            boundary_tolerance: characters two boundaries may be apart and still match.
            sentence_vectors: {chunk_id: [vector per sentence]} for semantic cohesion.
            chunk_vectors: {chunk_id: vector} for semantic chunk stickiness.

        Returns:
            A dict of metric name -> value (None = not measured) and "warnings".
        """
        check_size_limits(size_limits)
        documents = as_documents(document_text, documents)
        items, warnings = normalize_chunks(chunks, documents)

        scores = {"chunk_count": len(items)}
        scores.update(coverage_scores(items, documents))
        scores.update(size_scores(items, size_limits))
        scores.update(self._gold_scores(items, documents, expected_evidence, structural_blocks))
        scores["metadata_completeness"] = metadata_completeness(items, required_metadata)
        scores["metadata_accuracy"] = metadata_accuracy(items, expected_metadata)
        scores.update(lexical_scores(items))
        scores.update(semantic_scores(items, sentence_vectors, chunk_vectors))
        scores["boundary_f1"] = self._boundary_f1(items, expected_chunks, documents,
                                                  boundary_tolerance)
        scores["warnings"] = warnings
        return scores

    def _gold_scores(self, items, documents, expected_evidence, structural_blocks):
        """Check that gold evidence and structural blocks were not cut by a boundary."""
        self._need_documents(documents, expected_evidence, "expected_evidence")
        self._need_documents(documents, structural_blocks, "structural_blocks")

        evidence = evidence_spans(expected_evidence, documents)
        blocks = evidence_spans(structural_blocks, documents)
        return {"evidence_completeness": intact_rate(evidence, items),
                "evidence_fragmentation": fragmentation(evidence, items),
                "structural_integrity": intact_rate(blocks, items)}

    def _boundary_f1(self, items, expected_chunks, documents, tolerance):
        """Place the ideal chunks in the text and compare the split points."""
        if expected_chunks is None:
            return None
        self._need_documents(documents, expected_chunks, "expected_chunks")
        expected_items, _ = normalize_chunks(expected_chunks, documents)
        return boundary_f1(items, expected_items, tolerance)

    def _need_documents(self, documents, value, name):
        """Gold spans are places in the source text, so the text must be given."""
        if value and not documents:
            raise ValueError(f"{name} needs the source text: give document_text or documents")

evaluate(chunks, document_text=None, documents=None, expected_chunks=None, expected_evidence=None, structural_blocks=None, required_metadata=None, expected_metadata=None, size_limits=None, boundary_tolerance=0, sentence_vectors=None, chunk_vectors=None)

Score the chunks.

Parameters:

Name	Type	Description	Default
chunks		texts, or dicts with "id", "text", "start", "end", "doc_id", "metadata".	required
document_text		the source text of a single document.	None
documents		{doc_id: text} when there are several documents.	None
expected_chunks		the ideal chunks marked by a human (for boundary F1).	None
expected_evidence		gold quotes or {"start", "end"} spans that must stay whole.	None
structural_blocks		tables, lists or code that must stay whole (same format).	None
required_metadata		keys every chunk must have, e.g. ["source", "page"].	None
expected_metadata		{chunk_id: {key: value}} of correct metadata values.	None
size_limits		(min, max) words of a valid chunk.	None
boundary_tolerance		characters two boundaries may be apart and still match.	0
sentence_vectors		{chunk_id: [vector per sentence]} for semantic cohesion.	None
chunk_vectors		{chunk_id: vector} for semantic chunk stickiness.	None

Returns:

Type	Description
	A dict of metric name -> value (None = not measured) and "warnings".

Source code in src/rag_titan/evaluation/stages/chunking.py

def evaluate(self, chunks, document_text=None, documents=None, expected_chunks=None,
             expected_evidence=None, structural_blocks=None, required_metadata=None,
             expected_metadata=None, size_limits=None, boundary_tolerance=0,
             sentence_vectors=None, chunk_vectors=None):
    """
    Score the chunks.

    Args:
        chunks: texts, or dicts with "id", "text", "start", "end", "doc_id", "metadata".
        document_text: the source text of a single document.
        documents: {doc_id: text} when there are several documents.
        expected_chunks: the ideal chunks marked by a human (for boundary F1).
        expected_evidence: gold quotes or {"start", "end"} spans that must stay whole.
        structural_blocks: tables, lists or code that must stay whole (same format).
        required_metadata: keys every chunk must have, e.g. ["source", "page"].
        expected_metadata: {chunk_id: {key: value}} of correct metadata values.
        size_limits: (min, max) words of a valid chunk.
        boundary_tolerance: characters two boundaries may be apart and still match.
        sentence_vectors: {chunk_id: [vector per sentence]} for semantic cohesion.
        chunk_vectors: {chunk_id: vector} for semantic chunk stickiness.

    Returns:
        A dict of metric name -> value (None = not measured) and "warnings".
    """
    check_size_limits(size_limits)
    documents = as_documents(document_text, documents)
    items, warnings = normalize_chunks(chunks, documents)

    scores = {"chunk_count": len(items)}
    scores.update(coverage_scores(items, documents))
    scores.update(size_scores(items, size_limits))
    scores.update(self._gold_scores(items, documents, expected_evidence, structural_blocks))
    scores["metadata_completeness"] = metadata_completeness(items, required_metadata)
    scores["metadata_accuracy"] = metadata_accuracy(items, expected_metadata)
    scores.update(lexical_scores(items))
    scores.update(semantic_scores(items, sentence_vectors, chunk_vectors))
    scores["boundary_f1"] = self._boundary_f1(items, expected_chunks, documents,
                                              boundary_tolerance)
    scores["warnings"] = warnings
    return scores

EmbeddingEvaluator

Scores chunk and query vectors, and the results of a vector index.

Source code in src/rag_titan/evaluation/stages/embedding.py

class EmbeddingEvaluator:
    """Scores chunk and query vectors, and the results of a vector index."""

    def evaluate(self, vectors, query_vectors=None, relevant_ids=None, indexed_ids=None,
                 index_results=None, k=5):
        """
        Score the vectors.

        Args:
            vectors: {chunk_id: vector} of every chunk.
            query_vectors: {query_id: vector}.
            relevant_ids: {query_id: [gold chunk ids]}.
            indexed_ids: the chunk ids really stored in your vector index.
            index_results: {query_id: [chunk ids the index returned, best first]}.
            k: how many neighbours to compare.

        Returns:
            A dict of metric name -> value (None = not measured).
        """
        check_k(k)
        if not isinstance(vectors, dict):
            raise ValueError("vectors must be a dict of {chunk_id: vector}")

        scores = self._health(vectors)
        scores["index_coverage"] = None
        if indexed_ids is not None and vectors:
            scores["index_coverage"] = len(set(vectors) & set(indexed_ids)) / len(vectors)

        scores.update(self._search_scores(vectors, query_vectors or {}, relevant_ids or {},
                                          index_results, k))
        return scores

    def _health(self, vectors):
        """
        Basic checks on the stored vectors.

            zero_norm_rate         vectors of length 0 / all vectors (cannot be searched)
            duplicate_vector_rate  vectors equal to an earlier one / all vectors
        """
        sizes = {len(vector) for vector in vectors.values()}
        if len(sizes) > 1:
            raise ValueError(f"Vectors have different sizes: {sorted(sizes)}")

        norms = []
        for chunk_id, vector in vectors.items():
            if not all(is_number(value) for value in vector):
                raise ValueError(f"Vector of {chunk_id!r} has a value that is not a finite number")
            norms.append(math.sqrt(sum(value * value for value in vector)))

        count = len(vectors)
        distinct = len({tuple(vector) for vector in vectors.values()})
        return {
            "vector_count": count,
            "dimension": next(iter(sizes), None),
            "zero_norm_rate": sum(norm == 0 for norm in norms) / count if count else None,
            "duplicate_vector_rate": (count - distinct) / count if count else None,
            "mean_vector_norm": average(norms),
        }

    def _search_scores(self, vectors, query_vectors, relevant_ids, index_results, k):
        """
        Search every query by exact cosine and compare with the gold chunks and the index.

            query_topk_recall    gold chunks in the exact top K / gold chunks
            ann_recall_at_k      exact top K also returned by your index / exact top K
            positive_similarity  mean cos(query, gold chunk)
            negative_similarity  mean cos(query, other chunk)
            similarity_margin    positive - negative (higher = easier to separate)
        """
        recalls, ann_recalls, positive, negative = [], [], [], []
        for query_id, query_vector in query_vectors.items():
            scored = [(chunk_id, cosine(query_vector, vector))
                      for chunk_id, vector in vectors.items()]
            scored = [(chunk_id, score) for chunk_id, score in scored if score is not None]
            exact = [chunk_id for chunk_id, _ in
                     sorted(scored, key=lambda pair: pair[1], reverse=True)[:k]]

            gold = set(relevant_ids.get(query_id) or [])
            if gold:
                recalls.append(len(set(exact) & gold) / len(gold))
                positive += [score for chunk_id, score in scored if chunk_id in gold]
                negative += [score for chunk_id, score in scored if chunk_id not in gold]
            if index_results is not None and query_id in index_results and exact:
                returned = set(index_results[query_id][:k])
                ann_recalls.append(len(returned & set(exact)) / len(exact))

        return {
            "query_topk_recall": average(recalls),
            "ann_recall_at_k": average(ann_recalls),
            "positive_similarity": average(positive),
            "negative_similarity": average(negative),
            "similarity_margin": difference(average(positive), average(negative)),
        }

evaluate(vectors, query_vectors=None, relevant_ids=None, indexed_ids=None, index_results=None, k=5)

Score the vectors.

Parameters:

Name	Type	Description	Default
vectors		{chunk_id: vector} of every chunk.	required
query_vectors		{query_id: vector}.	None
relevant_ids		{query_id: [gold chunk ids]}.	None
indexed_ids		the chunk ids really stored in your vector index.	None
index_results		{query_id: [chunk ids the index returned, best first]}.	None
k		how many neighbours to compare.	5

Returns:

Type	Description
	A dict of metric name -> value (None = not measured).

Source code in src/rag_titan/evaluation/stages/embedding.py

def evaluate(self, vectors, query_vectors=None, relevant_ids=None, indexed_ids=None,
             index_results=None, k=5):
    """
    Score the vectors.

    Args:
        vectors: {chunk_id: vector} of every chunk.
        query_vectors: {query_id: vector}.
        relevant_ids: {query_id: [gold chunk ids]}.
        indexed_ids: the chunk ids really stored in your vector index.
        index_results: {query_id: [chunk ids the index returned, best first]}.
        k: how many neighbours to compare.

    Returns:
        A dict of metric name -> value (None = not measured).
    """
    check_k(k)
    if not isinstance(vectors, dict):
        raise ValueError("vectors must be a dict of {chunk_id: vector}")

    scores = self._health(vectors)
    scores["index_coverage"] = None
    if indexed_ids is not None and vectors:
        scores["index_coverage"] = len(set(vectors) & set(indexed_ids)) / len(vectors)

    scores.update(self._search_scores(vectors, query_vectors or {}, relevant_ids or {},
                                      index_results, k))
    return scores

EndToEndEvaluator

Scores the final answers, task outcomes and request telemetry.

Source code in src/rag_titan/evaluation/stages/end_to_end.py

class EndToEndEvaluator:
    """Scores the final answers, task outcomes and request telemetry."""

    def evaluate(self, queries):
        """
        Score every query that has an "answer".

        Args:
            queries: list of dicts with "id", "answer" and any of
                "expected_answer", "task_success", "latency_ms", "cost_usd",
                "total_tokens".

        Returns:
            A dict of metric name -> value (None = not measured):

                End-to-End P95 Latency = P95(t_request to answer)
                Task Success Rate      = #successful tasks / #labelled tasks
        """
        check_queries(queries)
        observed = [query for query in queries if "answer" in query]
        with_reference = [query for query in observed if query.get("expected_answer") is not None]
        outcomes = [query["task_success"] for query in observed
                    if query.get("task_success") is not None]
        times = self._numbers(observed, "latency_ms")

        return {
            "query_count": len(observed),
            "answer_exact_match": average([exact_match(query["answer"], query["expected_answer"])
                                           for query in with_reference]),
            "answer_f1": average([token_scores(query["answer"],
                                               query["expected_answer"])["answer_f1"]
                                  for query in with_reference]),
            "task_success_rate": average([float(bool(outcome)) for outcome in outcomes]),
            "p50_latency_ms": percentile(times, 50),
            "p95_latency_ms": percentile(times, 95),
            "mean_cost_usd": average(self._numbers(observed, "cost_usd")),
            "mean_total_tokens": average(self._numbers(observed, "total_tokens")),
        }

    def _numbers(self, queries, key):
        """Return the measured values of one telemetry key, checking they are valid."""
        values = []
        for query in queries:
            value = query.get(key)
            if value is None:
                continue
            if not is_number(value) or value < 0:
                raise ValueError(f"Query {query['id']!r}: {key!r} must be a number of 0 or more")
            values.append(value)
        return values

evaluate(queries)

Score every query that has an "answer".

Parameters:

Name	Type	Description	Default
queries		list of dicts with "id", "answer" and any of "expected_answer", "task_success", "latency_ms", "cost_usd", "total_tokens".	required

Returns:

Type	Description
	A dict of metric name -> value (None = not measured): End-to-End P95 Latency = P95(t_request to answer) Task Success Rate = #successful tasks / #labelled tasks

Source code in src/rag_titan/evaluation/stages/end_to_end.py

def evaluate(self, queries):
    """
    Score every query that has an "answer".

    Args:
        queries: list of dicts with "id", "answer" and any of
            "expected_answer", "task_success", "latency_ms", "cost_usd",
            "total_tokens".

    Returns:
        A dict of metric name -> value (None = not measured):

            End-to-End P95 Latency = P95(t_request to answer)
            Task Success Rate      = #successful tasks / #labelled tasks
    """
    check_queries(queries)
    observed = [query for query in queries if "answer" in query]
    with_reference = [query for query in observed if query.get("expected_answer") is not None]
    outcomes = [query["task_success"] for query in observed
                if query.get("task_success") is not None]
    times = self._numbers(observed, "latency_ms")

    return {
        "query_count": len(observed),
        "answer_exact_match": average([exact_match(query["answer"], query["expected_answer"])
                                       for query in with_reference]),
        "answer_f1": average([token_scores(query["answer"],
                                           query["expected_answer"])["answer_f1"]
                              for query in with_reference]),
        "task_success_rate": average([float(bool(outcome)) for outcome in outcomes]),
        "p50_latency_ms": percentile(times, 50),
        "p95_latency_ms": percentile(times, 95),
        "mean_cost_usd": average(self._numbers(observed, "cost_usd")),
        "mean_total_tokens": average(self._numbers(observed, "total_tokens")),
    }

GenerationEvaluator

Scores generated answers against reference answers and your labels.

Source code in src/rag_titan/evaluation/stages/generation.py

class GenerationEvaluator:
    """Scores generated answers against reference answers and your labels."""

    def evaluate(self, queries, chunks=None):
        """
        Score every query that has an "answer".

        Args:
            queries: list of dicts with "id", "answer" and any of:
                "expected_answer", "required_facts" (phrases), "citations" and
                "expected_citations" (source ids), "verified_claims" (list of
                {"supported": True/False}), "answerable" and "abstained",
                "task_success", "generation_latency_ms", and the context as
                "context" (text) or "context_ids".
            chunks: the chunks, needed only to turn "context_ids" into text.

        Returns:
            {"summary": the mean of each metric, "per_query": one dict per query}.
        """
        check_queries(queries)
        items, _ = normalize_chunks(chunks or [], {})
        texts = {item["id"]: item["text"] for item in items}

        per_query = [self._score_query(query, texts) for query in queries if "answer" in query]

        summary = {"query_count": len(per_query)}
        for key in SUMMARY_KEYS:
            summary[key] = average([scores[key] for scores in per_query])
        times = [scores["latency_ms"] for scores in per_query]
        summary["p50_latency_ms"] = percentile(times, 50)
        summary["p95_latency_ms"] = percentile(times, 95)
        return {"summary": summary, "per_query": per_query}

    def _score_query(self, query, texts):
        """Return every generation score of one query."""
        answer = query["answer"]
        if not isinstance(answer, str):
            raise ValueError(f"Query {query['id']!r}: 'answer' must be a text")

        scores = {"query_id": query["id"]}
        scores.update(self._reference_scores(answer, query.get("expected_answer")))
        scores["required_fact_coverage"] = required_fact_coverage(
            answer, query.get("required_facts"))
        scores.update(citation_scores(query.get("citations"), query.get("expected_citations")))
        scores.update(self._label_scores(query))

        context = self._context_text(query, texts)
        scores["context_grounding"] = context_grounding(answer, context)
        scores["unsupported_number_rate"] = unsupported_number_rate(answer, context)
        scores["latency_ms"] = query.get("generation_latency_ms")
        return scores

    def _reference_scores(self, answer, expected):
        """Compare the answer with the reference answer (None when there is none)."""
        if expected is None:
            return {key: None for key in ("exact_match", "answer_precision", "answer_recall",
                                          "answer_f1", "rouge_l_f1")}
        scores = {"exact_match": exact_match(answer, expected)}
        scores.update(token_scores(answer, expected))
        scores["rouge_l_f1"] = rouge_l_f1(answer, expected)
        return scores

    def _label_scores(self, query):
        """
        Scores that come from labels you supply.

            annotated_claim_support  claims marked supported / claims
            Abstention Accuracy      1 when "abstained" is right: the system
                                     refused an unanswerable question or
                                     answered an answerable one
        """
        claims = query.get("verified_claims")
        abstained, answerable = query.get("abstained"), query.get("answerable")

        support, abstention = None, None
        if claims:
            support = sum(bool(claim["supported"]) for claim in claims) / len(claims)
        if abstained is not None and answerable is not None:
            abstention = float(bool(abstained) != bool(answerable))

        success = query.get("task_success")
        return {"annotated_claim_support": support, "abstention_accuracy": abstention,
                "task_success": None if success is None else float(bool(success))}

    def _context_text(self, query, texts):
        """Return the context the LLM was given, or None when it is not known."""
        if query.get("context") is not None:
            return query["context"]

        found = [texts[chunk_id] for chunk_id in query.get("context_ids") or []
                 if chunk_id in texts]
        return " ".join(found) if found else None

evaluate(queries, chunks=None)

Score every query that has an "answer".

Parameters:

Name	Type	Description	Default
queries		list of dicts with "id", "answer" and any of: "expected_answer", "required_facts" (phrases), "citations" and "expected_citations" (source ids), "verified_claims" (list of {"supported": True/False}), "answerable" and "abstained", "task_success", "generation_latency_ms", and the context as "context" (text) or "context_ids".	required
chunks		the chunks, needed only to turn "context_ids" into text.	None

Returns:

Type	Description
	{"summary": the mean of each metric, "per_query": one dict per query}.

Source code in src/rag_titan/evaluation/stages/generation.py

def evaluate(self, queries, chunks=None):
    """
    Score every query that has an "answer".

    Args:
        queries: list of dicts with "id", "answer" and any of:
            "expected_answer", "required_facts" (phrases), "citations" and
            "expected_citations" (source ids), "verified_claims" (list of
            {"supported": True/False}), "answerable" and "abstained",
            "task_success", "generation_latency_ms", and the context as
            "context" (text) or "context_ids".
        chunks: the chunks, needed only to turn "context_ids" into text.

    Returns:
        {"summary": the mean of each metric, "per_query": one dict per query}.
    """
    check_queries(queries)
    items, _ = normalize_chunks(chunks or [], {})
    texts = {item["id"]: item["text"] for item in items}

    per_query = [self._score_query(query, texts) for query in queries if "answer" in query]

    summary = {"query_count": len(per_query)}
    for key in SUMMARY_KEYS:
        summary[key] = average([scores[key] for scores in per_query])
    times = [scores["latency_ms"] for scores in per_query]
    summary["p50_latency_ms"] = percentile(times, 50)
    summary["p95_latency_ms"] = percentile(times, 95)
    return {"summary": summary, "per_query": per_query}

RerankingEvaluator

Compares the order before ("retrieved_ids") and after ("reranked_ids") reranking.

Source code in src/rag_titan/evaluation/stages/reranking.py

class RerankingEvaluator:
    """Compares the order before ("retrieved_ids") and after ("reranked_ids") reranking."""

    def evaluate(self, queries, k=5):
        """
        Score every query that has "reranked_ids".

        Args:
            queries: list of dicts with "id", "reranked_ids" (best first), the
                gold labels ("expected_chunk_ids" or graded "relevance") and,
                to measure the gains, "retrieved_ids".
            k: how many results to score per query.

        Returns:
            {"summary": the mean of each metric, "per_query": one dict per query}.
            Gains are None for a query without "retrieved_ids".
        """
        check_k(k)
        check_queries(queries)
        per_query = [self._score_query(query, k) for query in queries if "reranked_ids" in query]

        summary = {"query_count": len(per_query)}
        for name, key in SUMMARY_KEYS.items():
            summary[name] = average([scores[key] for scores in per_query])
        return {"summary": summary, "per_query": per_query}

    def _score_query(self, query, k):
        """Score one query before and after reranking."""
        relevance = relevance_of(query)
        reranked = ranked_ids(query, "reranked_ids")
        after = ranking_scores(reranked, relevance, k)

        retrieved = ranked_ids(query, "retrieved_ids") if "retrieved_ids" in query else None
        before = {key: None for key in RANKING_KEYS}
        if retrieved is not None:
            before = ranking_scores(retrieved, relevance, k)

        scores = {"query_id": query["id"]}
        for name in ("ndcg", "mrr", "precision"):
            key = f"{name}_at_k"
            scores[f"before_{name}"] = before[key]
            scores[f"after_{name}"] = after[key]
            scores[f"{name}_gain"] = difference(after[key], before[key])

        scores["first_relevant_rank_change"] = self._rank_change(retrieved, reranked, relevance)
        scores["candidate_set_preserved"] = self._preserved(retrieved, reranked)
        return scores

    def _rank_change(self, retrieved, reranked, relevance):
        """
        How many places the first relevant chunk moved up (whole lists, not only top k).
        +2 means it went from rank 3 to rank 1; a negative number means it moved down.
        """
        if retrieved is None:
            return None
        before = first_relevant_rank(retrieved, relevance)
        after = first_relevant_rank(reranked, relevance)
        return difference(before, after)

    def _preserved(self, retrieved, reranked):
        """
        1.0 when the reranker returned exactly the chunks it was given (only reordered).
        A reranker that also cuts the list gives 0.0, which can be what you want.
        """
        if retrieved is None:
            return None
        return float(set(retrieved) == set(reranked))

evaluate(queries, k=5)

Score every query that has "reranked_ids".

Parameters:

Name	Type	Description	Default
queries		list of dicts with "id", "reranked_ids" (best first), the gold labels ("expected_chunk_ids" or graded "relevance") and, to measure the gains, "retrieved_ids".	required
k		how many results to score per query.	5

Returns:

Type	Description
	{"summary": the mean of each metric, "per_query": one dict per query}.
	Gains are None for a query without "retrieved_ids".

Source code in src/rag_titan/evaluation/stages/reranking.py

def evaluate(self, queries, k=5):
    """
    Score every query that has "reranked_ids".

    Args:
        queries: list of dicts with "id", "reranked_ids" (best first), the
            gold labels ("expected_chunk_ids" or graded "relevance") and,
            to measure the gains, "retrieved_ids".
        k: how many results to score per query.

    Returns:
        {"summary": the mean of each metric, "per_query": one dict per query}.
        Gains are None for a query without "retrieved_ids".
    """
    check_k(k)
    check_queries(queries)
    per_query = [self._score_query(query, k) for query in queries if "reranked_ids" in query]

    summary = {"query_count": len(per_query)}
    for name, key in SUMMARY_KEYS.items():
        summary[name] = average([scores[key] for scores in per_query])
    return {"summary": summary, "per_query": per_query}

RetrievalEvaluator

Scores the ranked chunk ids a retriever returned for each query.

Source code in src/rag_titan/evaluation/stages/retrieval.py

class RetrievalEvaluator:
    """Scores the ranked chunk ids a retriever returned for each query."""

    def evaluate(self, chunks, queries, k=5, document_text=None, documents=None):
        """
        Score every query that has "retrieved_ids".

        Args:
            chunks: all chunks of the index: texts, or dicts with "id", "text"
                and, when you have them, "start", "end" and "doc_id".
            queries: list of dicts with "id", "retrieved_ids" (best first) and
                the gold labels: "expected_chunk_ids" or graded "relevance",
                and optionally "expected_evidence" and "retrieval_latency_ms".
            k: how many results to score per query.
            document_text: the source text of a single document.
            documents: {doc_id: text} when there are several documents.

        Returns:
            {"summary": the mean of each metric, "per_query": one dict per
            query, "warnings": chunks that could not be placed in the text}.
            A metric is None when its inputs were not given.
        """
        check_k(k)
        check_queries(queries)
        documents = as_documents(document_text, documents)
        items, warnings = normalize_chunks(chunks, documents)

        by_id = {item["id"]: item for item in items}
        tokens = document_tokens(documents)
        per_query = [self._score_query(query, by_id, documents, tokens, k)
                     for query in queries if "retrieved_ids" in query]

        return {"summary": self._summary(per_query), "per_query": per_query,
                "warnings": warnings}

    def _score_query(self, query, by_id, documents, tokens, k):
        """Return the ranking and token scores of one query."""
        selected = ranked_ids(query, "retrieved_ids")[:k]
        scores = {"query_id": query["id"]}
        scores.update(ranking_scores(selected, relevance_of(query), k))
        scores.update(self._token_scores(query, selected, by_id, documents, tokens))
        scores["latency_ms"] = query.get("retrieval_latency_ms")
        scores["missing_chunk_ids"] = [cid for cid in selected if cid not in by_id]
        return scores

    def _token_scores(self, query, selected, by_id, documents, tokens):
        """
        Compare the words of the retrieved chunks (R) with the gold evidence words (G).

            Token Recall    = |R ∩ G| / |G|
            Token Precision = |R ∩ G| / |R|
            Token IoU       = |R ∩ G| / |R ∪ G|
            Complete Evidence Recall = #gold spans fully inside one chunk / #gold spans
        """
        evidence = evidence_spans(query.get("expected_evidence"), documents) if documents else []
        if not evidence:
            return {key: None for key in TOKEN_KEYS}

        spans = chunk_spans([by_id[cid] for cid in selected if cid in by_id])
        gold = covered_tokens(tokens, evidence)
        found = covered_tokens(tokens, spans)
        both = gold & found

        return {
            "token_recall": len(both) / len(gold) if gold else None,
            "token_precision": len(both) / len(found) if found else None,
            "token_iou": len(both) / len(gold | found) if gold | found else None,
            "complete_evidence_recall":
                sum(is_inside(span, spans) for span in evidence) / len(evidence),
        }

    def _summary(self, per_query):
        """Average every metric over the queries and add the latency percentiles."""
        summary = {"query_count": len(per_query)}
        for key in RANKING_KEYS + TOKEN_KEYS:
            summary[key] = average([scores[key] for scores in per_query])

        times = [scores["latency_ms"] for scores in per_query]
        summary["p50_latency_ms"] = percentile(times, 50)
        summary["p95_latency_ms"] = percentile(times, 95)
        return summary

evaluate(chunks, queries, k=5, document_text=None, documents=None)

Score every query that has "retrieved_ids".

Parameters:

Name	Type	Description	Default
chunks		all chunks of the index: texts, or dicts with "id", "text" and, when you have them, "start", "end" and "doc_id".	required
queries		list of dicts with "id", "retrieved_ids" (best first) and the gold labels: "expected_chunk_ids" or graded "relevance", and optionally "expected_evidence" and "retrieval_latency_ms".	required
k		how many results to score per query.	5
document_text		the source text of a single document.	None
documents		{doc_id: text} when there are several documents.	None

Returns:

Type	Description
	{"summary": the mean of each metric, "per_query": one dict per
	query, "warnings": chunks that could not be placed in the text}.
	A metric is None when its inputs were not given.

Source code in src/rag_titan/evaluation/stages/retrieval.py

def evaluate(self, chunks, queries, k=5, document_text=None, documents=None):
    """
    Score every query that has "retrieved_ids".

    Args:
        chunks: all chunks of the index: texts, or dicts with "id", "text"
            and, when you have them, "start", "end" and "doc_id".
        queries: list of dicts with "id", "retrieved_ids" (best first) and
            the gold labels: "expected_chunk_ids" or graded "relevance",
            and optionally "expected_evidence" and "retrieval_latency_ms".
        k: how many results to score per query.
        document_text: the source text of a single document.
        documents: {doc_id: text} when there are several documents.

    Returns:
        {"summary": the mean of each metric, "per_query": one dict per
        query, "warnings": chunks that could not be placed in the text}.
        A metric is None when its inputs were not given.
    """
    check_k(k)
    check_queries(queries)
    documents = as_documents(document_text, documents)
    items, warnings = normalize_chunks(chunks, documents)

    by_id = {item["id"]: item for item in items}
    tokens = document_tokens(documents)
    per_query = [self._score_query(query, by_id, documents, tokens, k)
                 for query in queries if "retrieved_ids" in query]

    return {"summary": self._summary(per_query), "per_query": per_query,
            "warnings": warnings}

Scoring: turn stage metrics into 0 to 100 scores, ratings and quality gates.

ScoreAggregator

Combines the stage metrics into stage scores, an overall score and a status.

Source code in src/rag_titan/evaluation/scoring/scorecard.py

class ScoreAggregator:
    """Combines the stage metrics into stage scores, an overall score and a status."""

    def __init__(self, stage_weights=None, metric_weights=None, targets=None,
                 quality_gates=None, stages=None, min_coverage=0.75):
        """
        Args:
            stage_weights: {stage: weight} to change how much a stage counts.
            metric_weights: {stage: {metric: weight}}; weight 0 switches a metric off.
            targets: {stage: {metric: target}} for sizes, latency and cost, e.g.
                {"end_to_end": {"p95_latency_ms": 2000}}.
            quality_gates: {stage: {metric: minimum}} replacing the default gates.
            stages: the stages to score, e.g. ["retrieval"]; default is all.
            min_coverage: below this measured share the score is "Provisional".
        """
        self.stages = list(STAGE_WEIGHTS) if stages is None else list(stages)
        unknown = set(self.stages) - set(STAGE_WEIGHTS)
        if unknown or not self.stages or len(self.stages) != len(set(self.stages)):
            raise ValueError(f"stages must be unique names from {list(STAGE_WEIGHTS)}, "
                             f"got {self.stages}")
        if not is_number(min_coverage) or not 0 <= min_coverage <= 1:
            raise ValueError("min_coverage must be between 0 and 1")

        self.stage_weights = {stage: STAGE_WEIGHTS[stage] for stage in self.stages}
        self.stage_weights.update(stage_weights or {})
        self.metric_weights = metric_weights or {}
        self.targets = targets or {}
        self.quality_gates = QUALITY_GATES if quality_gates is None else quality_gates
        self.min_coverage = min_coverage
        self._check_weights()

    def _check_weights(self):
        """Raise a clear error for a weight that is unknown or not a valid number."""
        for stage, weight in self.stage_weights.items():
            if stage not in self.stages or not is_number(weight) or weight <= 0:
                raise ValueError(f"Stage weight of {stage!r} must be a number above 0 "
                                 "for a scored stage")

        for stage, weights in self.metric_weights.items():
            known = set(METRICS.get(stage, {})) | set(TARGET_METRICS.get(stage, {}))
            for name, weight in weights.items():
                if name not in known or not is_number(weight) or weight < 0:
                    raise ValueError(f"Invalid metric weight for {stage}.{name}")

    def evaluate(self, report):
        """
        Score a report from RAGEvaluator (or a dict of {stage: stage result}).

        Returns:
            {"overall": score, rating, status, coverage, gate failures and the
            stages to review, "stages": one entry per stage, "stage_weights", "method"}.
        """
        cards, weighted, measured = {}, 0.0, 0.0
        for stage in self.stages:
            rules = stage_rules(stage, self.metric_weights, self.targets)
            card, score, coverage = score_stage(stage, report.get(stage) or {}, rules,
                                                self.targets.get(stage, {}), self.min_coverage)
            cards[stage] = card
            measured += self.stage_weights[stage] * coverage
            weighted += self.stage_weights[stage] * coverage * (score or 0.0)

        score = weighted / measured if measured else None
        coverage = measured / sum(self.stage_weights.values())
        return {"overall": self._overall(score, coverage, cards, self._gates(report)),
                "stages": cards, "stage_weights": dict(self.stage_weights), "method": METHOD}

    def _gates(self, report):
        """Return the quality gates that failed: measured metrics below their minimum."""
        failures = []
        for stage, gates in self.quality_gates.items():
            if stage not in self.stages:
                continue
            result = report.get(stage) or {}
            values = result.get("summary", result)
            for metric, minimum in gates.items():
                value = values.get(metric)
                if is_number(value) and value < minimum:
                    failures.append({
                        "stage": stage, "metric": metric, "observed": value, "minimum": minimum,
                        "message": f"{STAGE_NAMES[stage][0]}: {metric.replace('_', ' ')} "
                                   "is below the recommended minimum."})
        return failures

    def _overall(self, score, coverage, cards, gates):
        """Build the "overall" entry: score, rating, status and what to review first."""
        label, message = rating(None if score is None else score * 100)
        provisional = coverage < self.min_coverage

        if score is None:
            status = "Not Evaluated"
        elif gates:
            status = "Action Required"
            message += " One or more critical quality checks failed."
        elif provisional:
            status = "Provisional"
        else:
            status = "Evaluated"
            message += " Review individual stages before deciding whether to deploy."
        if provisional and score is not None:
            message += " This is a provisional score because evaluation data is incomplete."

        scored = [stage for stage in self.stages if cards[stage]["score"] is not None]
        weakest = sorted(scored, key=lambda stage: cards[stage]["score"])[:3]
        return {
            "score": None if score is None else round(score * 100, 2),
            "score_0_1": None if score is None else round(score, 4),
            "rating": label, "status": status, "message": message,
            "evaluation_coverage": round(coverage, 4), "is_provisional": provisional,
            "evaluated_stages": len(scored), "total_stages": len(self.stages),
            "quality_gate_failures": gates,
            "areas_to_review": [{"stage": stage, "label": cards[stage]["name"],
                                 "score": cards[stage]["score"],
                                 "improvement_areas": cards[stage]["improvement_areas"]}
                                for stage in weakest],
        }

__init__(stage_weights=None, metric_weights=None, targets=None, quality_gates=None, stages=None, min_coverage=0.75)

Parameters:

Name	Type	Description	Default
stage_weights		{stage: weight} to change how much a stage counts.	None
metric_weights		{stage: {metric: weight}}; weight 0 switches a metric off.	None
targets		{stage: {metric: target}} for sizes, latency and cost, e.g. {"end_to_end": {"p95_latency_ms": 2000}}.	None
quality_gates		{stage: {metric: minimum}} replacing the default gates.	None
stages		the stages to score, e.g. ["retrieval"]; default is all.	None
min_coverage		below this measured share the score is "Provisional".	0.75

Source code in src/rag_titan/evaluation/scoring/scorecard.py

def __init__(self, stage_weights=None, metric_weights=None, targets=None,
             quality_gates=None, stages=None, min_coverage=0.75):
    """
    Args:
        stage_weights: {stage: weight} to change how much a stage counts.
        metric_weights: {stage: {metric: weight}}; weight 0 switches a metric off.
        targets: {stage: {metric: target}} for sizes, latency and cost, e.g.
            {"end_to_end": {"p95_latency_ms": 2000}}.
        quality_gates: {stage: {metric: minimum}} replacing the default gates.
        stages: the stages to score, e.g. ["retrieval"]; default is all.
        min_coverage: below this measured share the score is "Provisional".
    """
    self.stages = list(STAGE_WEIGHTS) if stages is None else list(stages)
    unknown = set(self.stages) - set(STAGE_WEIGHTS)
    if unknown or not self.stages or len(self.stages) != len(set(self.stages)):
        raise ValueError(f"stages must be unique names from {list(STAGE_WEIGHTS)}, "
                         f"got {self.stages}")
    if not is_number(min_coverage) or not 0 <= min_coverage <= 1:
        raise ValueError("min_coverage must be between 0 and 1")

    self.stage_weights = {stage: STAGE_WEIGHTS[stage] for stage in self.stages}
    self.stage_weights.update(stage_weights or {})
    self.metric_weights = metric_weights or {}
    self.targets = targets or {}
    self.quality_gates = QUALITY_GATES if quality_gates is None else quality_gates
    self.min_coverage = min_coverage
    self._check_weights()

evaluate(report)

Score a report from RAGEvaluator (or a dict of {stage: stage result}).

Returns:

Type	Description
	{"overall": score, rating, status, coverage, gate failures and the
	stages to review, "stages": one entry per stage, "stage_weights", "method"}.

Source code in src/rag_titan/evaluation/scoring/scorecard.py

def evaluate(self, report):
    """
    Score a report from RAGEvaluator (or a dict of {stage: stage result}).

    Returns:
        {"overall": score, rating, status, coverage, gate failures and the
        stages to review, "stages": one entry per stage, "stage_weights", "method"}.
    """
    cards, weighted, measured = {}, 0.0, 0.0
    for stage in self.stages:
        rules = stage_rules(stage, self.metric_weights, self.targets)
        card, score, coverage = score_stage(stage, report.get(stage) or {}, rules,
                                            self.targets.get(stage, {}), self.min_coverage)
        cards[stage] = card
        measured += self.stage_weights[stage] * coverage
        weighted += self.stage_weights[stage] * coverage * (score or 0.0)

    score = weighted / measured if measured else None
    coverage = measured / sum(self.stage_weights.values())
    return {"overall": self._overall(score, coverage, cards, self._gates(report)),
            "stages": cards, "stage_weights": dict(self.stage_weights), "method": METHOD}

rating(score)

Return (rating, meaning) for a 0 to 100 score; None means nothing was measured.

Source code in src/rag_titan/evaluation/scoring/policy.py

def rating(score):
    """Return (rating, meaning) for a 0 to 100 score; None means nothing was measured."""
    if score is None:
        return "Not Evaluated", "Not enough applicable measurements to calculate a score."
    return next((name, meaning) for minimum, name, meaning in RATINGS if score >= minimum)

Reports: save evaluation reports as JSON or Markdown and compare them over time.

ReportManager

One place for everything you do with a finished report.

Source code in src/rag_titan/evaluation/reports/manager.py

class ReportManager:
    """One place for everything you do with a finished report."""

    def save_json(self, report, path):
        """Write the report as JSON (folders are created) and return the path."""
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8")
        return path

    def load_json(self, path):
        """Read a report saved with save_json()."""
        path = Path(path)
        if not path.is_file():
            raise FileNotFoundError(f"Report file not found: {path}")
        try:
            report = json.loads(path.read_text(encoding="utf-8"))
        except json.JSONDecodeError as error:
            raise ValueError(f"{path} is not valid JSON (line {error.lineno}): {error.msg}")
        if not isinstance(report, dict):
            raise ValueError(f"{path} does not hold a report (a JSON object)")
        return report

    def markdown(self, report, title="RAG Pipeline Evaluation", generated=None):
        """Return the report as Markdown text (see to_markdown)."""
        return to_markdown(report, title, generated)

    def save_markdown(self, report, path, title="RAG Pipeline Evaluation"):
        """Write the report as Markdown (folders are created) and return the path."""
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(self.markdown(report, title), encoding="utf-8")
        return path

    def flatten(self, report):
        """Return every numeric stage metric as {"stage.metric": value}."""
        return flatten(report)

    def regressions(self, report, baseline, tolerance=0.02, latency_share=LATENCY_SHARE,
                    latency_min_ms=LATENCY_MIN_MS):
        """List the metrics that got worse than in the baseline (see find_regressions)."""
        return find_regressions(report, baseline, tolerance, latency_share, latency_min_ms)

    def trends(self, reports):
        """Return the first-to-last change of every metric over reports, oldest first."""
        return trends(reports)

    def failed_queries(self, report):
        """Return the queries that failed a check, earliest stage first."""
        return failed_queries(report)

failed_queries(report)

Return the queries that failed a check, earliest stage first.

Source code in src/rag_titan/evaluation/reports/manager.py

def failed_queries(self, report):
    """Return the queries that failed a check, earliest stage first."""
    return failed_queries(report)

flatten(report)

Return every numeric stage metric as {"stage.metric": value}.

Source code in src/rag_titan/evaluation/reports/manager.py

def flatten(self, report):
    """Return every numeric stage metric as {"stage.metric": value}."""
    return flatten(report)

load_json(path)

Read a report saved with save_json().

Source code in src/rag_titan/evaluation/reports/manager.py

def load_json(self, path):
    """Read a report saved with save_json()."""
    path = Path(path)
    if not path.is_file():
        raise FileNotFoundError(f"Report file not found: {path}")
    try:
        report = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as error:
        raise ValueError(f"{path} is not valid JSON (line {error.lineno}): {error.msg}")
    if not isinstance(report, dict):
        raise ValueError(f"{path} does not hold a report (a JSON object)")
    return report

markdown(report, title='RAG Pipeline Evaluation', generated=None)

Return the report as Markdown text (see to_markdown).

Source code in src/rag_titan/evaluation/reports/manager.py

def markdown(self, report, title="RAG Pipeline Evaluation", generated=None):
    """Return the report as Markdown text (see to_markdown)."""
    return to_markdown(report, title, generated)

regressions(report, baseline, tolerance=0.02, latency_share=LATENCY_SHARE, latency_min_ms=LATENCY_MIN_MS)

List the metrics that got worse than in the baseline (see find_regressions).

Source code in src/rag_titan/evaluation/reports/manager.py

def regressions(self, report, baseline, tolerance=0.02, latency_share=LATENCY_SHARE,
                latency_min_ms=LATENCY_MIN_MS):
    """List the metrics that got worse than in the baseline (see find_regressions)."""
    return find_regressions(report, baseline, tolerance, latency_share, latency_min_ms)

save_json(report, path)

Write the report as JSON (folders are created) and return the path.

Source code in src/rag_titan/evaluation/reports/manager.py

def save_json(self, report, path):
    """Write the report as JSON (folders are created) and return the path."""
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8")
    return path

save_markdown(report, path, title='RAG Pipeline Evaluation')

Write the report as Markdown (folders are created) and return the path.

Source code in src/rag_titan/evaluation/reports/manager.py

def save_markdown(self, report, path, title="RAG Pipeline Evaluation"):
    """Write the report as Markdown (folders are created) and return the path."""
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(self.markdown(report, title), encoding="utf-8")
    return path

trends(reports)

Return the first-to-last change of every metric over reports, oldest first.

Source code in src/rag_titan/evaluation/reports/manager.py

def trends(self, reports):
    """Return the first-to-last change of every metric over reports, oldest first."""
    return trends(reports)

evaluate_file(input_path, output_path='rag_report.json', k=5, baseline=None)

Evaluate a data file and save the report as JSON and Markdown.

Parameters:

Name	Type	Description	Default
input_path		the JSON data file.	required
output_path		where to write the JSON report; the Markdown report gets the same name ending in .md.	'rag_report.json'
k		how many results to score in the ranking metrics.	5
baseline		path of an earlier JSON report; the metrics that got worse are added to the report as "regressions".	None

Returns:

Type	Description
	The report dict.

Source code in src/rag_titan/evaluation/reports/run.py

def evaluate_file(input_path, output_path="rag_report.json", k=5, baseline=None):
    """
    Evaluate a data file and save the report as JSON and Markdown.

    Args:
        input_path: the JSON data file.
        output_path: where to write the JSON report; the Markdown report gets
            the same name ending in .md.
        k: how many results to score in the ranking metrics.
        baseline: path of an earlier JSON report; the metrics that got worse
            are added to the report as "regressions".

    Returns:
        The report dict.
    """
    manager = ReportManager()
    report = RAGEvaluator(k=k).evaluate(**load_dataset(input_path))
    if baseline:
        report["regressions"] = manager.regressions(report, manager.load_json(baseline))

    manager.save_json(report, output_path)
    manager.save_markdown(report, Path(output_path).with_suffix(".md"))
    return report

failed_queries(report)

Return [{"query_id", "stage", "reason"}] for the queries that failed a check.

Parameters:

Name	Type	Description	Default
report		a report from RAGEvaluator.	required

Source code in src/rag_titan/evaluation/reports/failures.py

def failed_queries(report):
    """
    Return [{"query_id", "stage", "reason"}] for the queries that failed a check.

    Args:
        report: a report from RAGEvaluator.
    """
    found = []
    for stage, key, bad_value, reason in RULES:
        for scores in _per_query(report, stage):
            if scores.get(key) == bad_value:
                found.append({"query_id": scores["query_id"], "stage": stage, "reason": reason})

    for scores in _per_query(report, "generation"):
        answer_f1 = scores.get("answer_f1")
        if answer_f1 is not None and answer_f1 < LOW_ANSWER_F1:
            found.append({"query_id": scores["query_id"], "stage": "generation",
                          "reason": f"answer differs from the reference (F1 {answer_f1:.2f})"})

    order = [stage for stage, _, _, _ in RULES]
    return sorted(found, key=lambda item: order.index(item["stage"]))

find_regressions(report, baseline, tolerance=0.02, latency_share=LATENCY_SHARE, latency_min_ms=LATENCY_MIN_MS)

List the metrics that got worse than in the baseline report.

Parameters:

Name	Type	Description	Default
report		the new report.	required
baseline		an earlier report to compare with.	required
tolerance		how much worse still counts as noise. For scores it is an absolute change (0.02 = 2 points of 100); for latency, cost and tokens a relative one (0.02 = 2% slower or more expensive).	0.02
latency_share		a latency must also be this share slower (0.10 = 10%).	LATENCY_SHARE
latency_min_ms		... and this many milliseconds slower. Set both to 0 to count every slower timing.	LATENCY_MIN_MS

Returns:

Type	Description
	A list of {"metric", "baseline", "current", "delta"}, in pipeline order.

Source code in src/rag_titan/evaluation/reports/compare.py

def find_regressions(report, baseline, tolerance=0.02, latency_share=LATENCY_SHARE,
                     latency_min_ms=LATENCY_MIN_MS):
    """
    List the metrics that got worse than in the baseline report.

    Args:
        report: the new report.
        baseline: an earlier report to compare with.
        tolerance: how much worse still counts as noise. For scores it is an
            absolute change (0.02 = 2 points of 100); for latency, cost and
            tokens a relative one (0.02 = 2% slower or more expensive).
        latency_share: a latency must also be this share slower (0.10 = 10%).
        latency_min_ms: ... and this many milliseconds slower. Set both to 0
            to count every slower timing.

    Returns:
        A list of {"metric", "baseline", "current", "delta"}, in pipeline order.
    """
    for label, value in (("tolerance", tolerance), ("latency_share", latency_share),
                         ("latency_min_ms", latency_min_ms)):
        if not is_number(value) or value < 0:
            raise ValueError(f"{label} must be a number of 0 or more, got {value!r}")

    current, previous = flatten(report), flatten(baseline)
    found = []
    for name, value in current.items():
        direction = better_direction(name)
        if name not in previous or direction is None:
            continue

        old = previous[name]
        worse_by = value - old if direction == "lower" else old - value
        metric = name.partition(".")[2]
        allowed = tolerance * abs(old) if metric in TELEMETRY else tolerance
        if metric.endswith("latency_ms"):
            allowed = max(allowed, latency_share * abs(old), latency_min_ms)
        if worse_by > allowed:
            found.append({"metric": name, "baseline": old, "current": value,
                          "delta": value - old})
    return found

flatten(report)

Return every numeric stage metric of a report as {"stage.metric": value}.

Source code in src/rag_titan/evaluation/reports/compare.py

def flatten(report):
    """Return every numeric stage metric of a report as {"stage.metric": value}."""
    if not isinstance(report, dict):
        raise ValueError("A report must be a dict")

    flat = {}
    for stage in STAGE_WEIGHTS:
        result = report.get(stage)
        if not isinstance(result, dict):
            continue
        for name, value in result.get("summary", result).items():
            if is_number(value):
                flat[f"{stage}.{name}"] = value
    return flat

summary_text(report)

Return the lines printed after an evaluation: score, status, first failure, the queries to look at, regressions and what was not measured.

Source code in src/rag_titan/evaluation/reports/run.py

def summary_text(report):
    """
    Return the lines printed after an evaluation: score, status, first failure,
    the queries to look at, regressions and what was not measured.
    """
    overall = report["scorecard"]["overall"]
    score = "N/A" if overall["score"] is None else f"{overall['score']:.1f}/100"
    lines = [
        f"Overall score: {score} ({overall['rating']})",
        f"Status: {overall['status']} | Measurement coverage: "
        f"{overall['evaluation_coverage']:.0%}",
        f"First failing stage: {report['root_causes']['first_failing_stage'] or 'none'}",
    ]
    # a wrong answer must be visible here, not only at the end of the .md report
    lines += _section("Queries to look at:", queries_to_look_at(report))
    if "regressions" in report:
        names = ", ".join(item["metric"] for item in report["regressions"]) or "none"
        lines.append(f"Regressions against the baseline: {names}")
    lines += _section("Not measured (null in the report) - to measure it:", not_measured(report))
    return "\n".join(lines)

to_markdown(report, title='RAG Pipeline Evaluation', generated=None)

Return the report as Markdown text.

Parameters:

Name	Type	Description	Default
report		a report from RAGEvaluator. A "regressions" entry is shown when present.	required
title		the heading of the report.	'RAG Pipeline Evaluation'
generated		the timestamp text; default is the current UTC time.	None

Source code in src/rag_titan/evaluation/reports/markdown.py

def to_markdown(report, title="RAG Pipeline Evaluation", generated=None):
    """
    Return the report as Markdown text.

    Args:
        report: a report from RAGEvaluator. A "regressions" entry is shown when present.
        title: the heading of the report.
        generated: the timestamp text; default is the current UTC time.
    """
    if not isinstance(report, dict):
        raise ValueError("A report must be a dict")
    generated = generated or datetime.now(timezone.utc).isoformat(timespec="seconds")

    lines = [f"# {title}", "", f"Generated: {generated}", ""]
    if report.get("scorecard"):
        lines += _scorecard_lines(report["scorecard"])
    if "regressions" in report:
        lines += _regression_lines(report["regressions"])
    for stage in STAGE_WEIGHTS:
        if isinstance(report.get(stage), dict):
            lines += _stage_lines(stage, report[stage])
    lines += _failure_lines(failed_queries(report))
    lines += _hint_lines(report.get("root_causes") or {})
    return "\n".join(lines) + "\n"

trends(reports)

Return the change of every metric from the first to the last report.

Parameters:

Name	Type	Description	Default
reports		report dicts, oldest first.	required

Returns:

Type	Description
	{"stage.metric": {"first", "last", "change", "improved", "runs"}} for the
	metrics present in every report. "improved" is None when neither
	direction is better. Empty with fewer than two reports.

Source code in src/rag_titan/evaluation/reports/compare.py

def trends(reports):
    """
    Return the change of every metric from the first to the last report.

    Args:
        reports: report dicts, oldest first.

    Returns:
        {"stage.metric": {"first", "last", "change", "improved", "runs"}} for the
        metrics present in every report. "improved" is None when neither
        direction is better. Empty with fewer than two reports.
    """
    history = [flatten(report) for report in reports]
    if len(history) < 2:
        return {}

    shared = set(history[0]).intersection(*history[1:])
    result = {}
    for name in sorted(shared):
        change = history[-1][name] - history[0][name]
        direction = better_direction(name)
        improved = None
        if direction and change:
            improved = (change < 0) == (direction == "lower")
        result[name] = {"first": history[0][name], "last": history[-1][name],
                        "change": change, "improved": improved, "runs": len(history)}
    return result

Generators that write the final answer.

BaseGenerator

Writes the final answer from the question and the chosen chunks.

Parameters:

Name	Type	Description	Default
llm		any BaseLLM.	required
instructions		the system prompt for the LLM.	required

Source code in src/rag_titan/generation/base_generator.py

class BaseGenerator:
    """
    Writes the final answer from the question and the chosen chunks.

    Args:
        llm: any BaseLLM.
        instructions: the system prompt for the LLM.
    """

    def __init__(self, llm, instructions):
        self.llm = llm
        self.instructions = instructions

    def generate(self, query, chunks):
        """Return the final answer text for the question."""
        raise NotImplementedError

    def _build_user_input(self, query, chunks):
        """Put the context and the question into one user message."""
        context = build_context(chunks)
        return f"CONTEXT:\n{context}\n\nQUESTION:\n{query}"

generate(query, chunks)

Return the final answer text for the question.

Source code in src/rag_titan/generation/base_generator.py

def generate(self, query, chunks):
    """Return the final answer text for the question."""
    raise NotImplementedError

GroundedGenerator

Bases: BaseGenerator

Answers only from the given chunks and cites their ids.

Parameters:

Name	Type	Description	Default
llm		any BaseLLM.	required
instructions		system prompt, defaults to GROUNDED_ANSWER_PROMPT.	GROUNDED_ANSWER_PROMPT

Source code in src/rag_titan/generation/grounded_generator.py

class GroundedGenerator(BaseGenerator):
    """
    Answers only from the given chunks and cites their ids.

    Args:
        llm: any BaseLLM.
        instructions: system prompt, defaults to GROUNDED_ANSWER_PROMPT.
    """

    def __init__(self, llm, instructions=GROUNDED_ANSWER_PROMPT):
        super().__init__(llm, instructions)

    def generate(self, query, chunks):
        """Return an answer with citations, or NO_ANSWER when unsure."""
        if not chunks:
            return NO_ANSWER

        user_input = self._build_user_input(query, chunks)
        answer = self.llm.ask(self.instructions, user_input)

        return answer or NO_ANSWER

generate(query, chunks)

Return an answer with citations, or NO_ANSWER when unsure.

Source code in src/rag_titan/generation/grounded_generator.py

def generate(self, query, chunks):
    """Return an answer with citations, or NO_ANSWER when unsure."""
    if not chunks:
        return NO_ANSWER

    user_input = self._build_user_input(query, chunks)
    answer = self.llm.ask(self.instructions, user_input)

    return answer or NO_ANSWER

build_context(chunks)

Join chunks into one text block the LLM can read and cite.

Each chunk looks like

[chunk_3] Source: handbook.pdf, page 2 ...chunk text...

Source code in src/rag_titan/generation/context_builder.py

def build_context(chunks):
    """
    Join chunks into one text block the LLM can read and cite.

    Each chunk looks like:
        [chunk_3] Source: handbook.pdf, page 2
        ...chunk text...
    """
    return "\n\n".join(f"{describe_source(chunk)}\n{chunk['text']}" for chunk in chunks)

Pipelines that prepare documents for search.

IndexingPipeline

Prepares files so they can be searched.

Parameters:

Name	Type	Description	Default
chunker		a BaseChunker.	required
embedder		a BaseEmbedder.	required
store		a BaseStore.	required

Source code in src/rag_titan/pipelines/indexing_pipeline.py

class IndexingPipeline:
    """
    Prepares files so they can be searched.

    Args:
        chunker: a BaseChunker.
        embedder: a BaseEmbedder.
        store: a BaseStore.
    """

    def __init__(self, chunker, embedder, store):
        self.chunker = chunker
        self.embedder = embedder
        self.store = store

    def run(self, file_paths, reset=True):
        """
        Index the files.

        Args:
            file_paths: list of file paths to index.
            reset: when True, clear the store before adding new chunks.

        Returns:
            The number of chunks stored.
        """
        documents = self.load_documents(file_paths)

        chunks = self.chunker.chunk(documents)
        logger.info("Created %d chunks", len(chunks))

        if not chunks:
            logger.warning("No chunks created, nothing to index")
            return 0

        vectors = self.embedder.embed_many([chunk["text"] for chunk in chunks])

        if reset:
            self.store.clear()

        self.store.add(chunks, vectors)
        logger.info("Stored %d chunks in the vector store", self.store.count())

        return len(chunks)

    def load_documents(self, file_paths):
        """Load every file and skip the ones without any text."""
        documents = []

        for path in file_paths:
            document = load_document(path)

            if not document["text"].strip():
                logger.warning("Skipping empty document: %s", document["source"])
                continue

            word_count = len(document["text"].split())
            logger.info("Loaded %s (%d words)", document["source"], word_count)
            documents.append(document)

        return documents

load_documents(file_paths)

Load every file and skip the ones without any text.

Source code in src/rag_titan/pipelines/indexing_pipeline.py

def load_documents(self, file_paths):
    """Load every file and skip the ones without any text."""
    documents = []

    for path in file_paths:
        document = load_document(path)

        if not document["text"].strip():
            logger.warning("Skipping empty document: %s", document["source"])
            continue

        word_count = len(document["text"].split())
        logger.info("Loaded %s (%d words)", document["source"], word_count)
        documents.append(document)

    return documents

run(file_paths, reset=True)

Index the files.

Parameters:

Name	Type	Description	Default
file_paths		list of file paths to index.	required
reset		when True, clear the store before adding new chunks.	True

Returns:

Type	Description
	The number of chunks stored.

Source code in src/rag_titan/pipelines/indexing_pipeline.py

def run(self, file_paths, reset=True):
    """
    Index the files.

    Args:
        file_paths: list of file paths to index.
        reset: when True, clear the store before adding new chunks.

    Returns:
        The number of chunks stored.
    """
    documents = self.load_documents(file_paths)

    chunks = self.chunker.chunk(documents)
    logger.info("Created %d chunks", len(chunks))

    if not chunks:
        logger.warning("No chunks created, nothing to index")
        return 0

    vectors = self.embedder.embed_many([chunk["text"] for chunk in chunks])

    if reset:
        self.store.clear()

    self.store.add(chunks, vectors)
    logger.info("Stored %d chunks in the vector store", self.store.count())

    return len(chunks)

Tool-calling agents.

BaseAgent

A simple agent that lets an LLM call Python functions.

Parameters:

Name	Type	Description	Default
llm		any BaseLLM that supports tool calling.	required
system_prompt		instructions that describe the agent's job.	required
tools		list of plain Python functions the LLM may call.	required
max_steps		stop after this many think/tool rounds.	5

Source code in src/rag_titan/agents/base_agent.py

class BaseAgent:
    """
    A simple agent that lets an LLM call Python functions.

    Args:
        llm: any BaseLLM that supports tool calling.
        system_prompt: instructions that describe the agent's job.
        tools: list of plain Python functions the LLM may call.
        max_steps: stop after this many think/tool rounds.
    """

    def __init__(self, llm, system_prompt, tools, max_steps=5):
        self.llm = llm
        self.system_prompt = system_prompt
        self.tools = tools
        self.tools_by_name = {tool.__name__: tool for tool in tools}
        self.max_steps = max_steps
        self.reset()

    def reset(self):
        """Forget the conversation and start fresh."""
        self.messages = [{"role": "system", "content": self.system_prompt}]

    def run(self, query):
        """
        Answer one user message, calling tools as needed.

        Args:
            query: the user's message.

        Returns:
            The agent's final answer text.
        """
        self.messages.append({"role": "user", "content": query})

        for step in range(1, self.max_steps + 1):
            logger.info("Agent step %d: thinking...", step)
            response = self.llm.chat(self.messages, tools=self.tools)
            self.messages.append(response["message"])

            # no tool calls means the LLM has given its final answer
            if not response["tool_calls"]:
                return response["content"]

            for tool_call in response["tool_calls"]:
                result = self._run_tool(tool_call)
                self.messages.append(self.llm.tool_result_message(tool_call, result))

        return "Maximum steps reached without a final answer."

    def _run_tool(self, tool_call):
        """Run one tool safely. Errors are returned as text to the LLM."""
        name = tool_call["name"]
        arguments = tool_call["arguments"]
        tool = self.tools_by_name.get(name)

        if tool is None:
            result = f"Error: unknown tool '{name}'"
        else:
            try:
                result = tool(**arguments)
            except Exception as error:
                result = f"Error while running {name}: {error}"

        logger.info("Tool %s(%s) = %s", name, arguments, result)
        return result

reset()

Forget the conversation and start fresh.

Source code in src/rag_titan/agents/base_agent.py

def reset(self):
    """Forget the conversation and start fresh."""
    self.messages = [{"role": "system", "content": self.system_prompt}]

run(query)

Answer one user message, calling tools as needed.

Parameters:

Name	Type	Description	Default
query		the user's message.	required

Returns:

Type	Description
	The agent's final answer text.

Source code in src/rag_titan/agents/base_agent.py

def run(self, query):
    """
    Answer one user message, calling tools as needed.

    Args:
        query: the user's message.

    Returns:
        The agent's final answer text.
    """
    self.messages.append({"role": "user", "content": query})

    for step in range(1, self.max_steps + 1):
        logger.info("Agent step %d: thinking...", step)
        response = self.llm.chat(self.messages, tools=self.tools)
        self.messages.append(response["message"])

        # no tool calls means the LLM has given its final answer
        if not response["tool_calls"]:
            return response["content"]

        for tool_call in response["tool_calls"]:
            result = self._run_tool(tool_call)
            self.messages.append(self.llm.tool_result_message(tool_call, result))

    return "Maximum steps reached without a final answer."

MathAgent

Bases: BaseAgent

Agent that uses add, subtract, multiply and divide tools.

Parameters:

Name	Type	Description	Default
llm		a BaseLLM whose model supports tool calling.	required
max_steps		stop after this many think/tool rounds.	5

Source code in src/rag_titan/agents/math_agent.py

class MathAgent(BaseAgent):
    """
    Agent that uses add, subtract, multiply and divide tools.

    Args:
        llm: a BaseLLM whose model supports tool calling.
        max_steps: stop after this many think/tool rounds.
    """

    def __init__(self, llm, max_steps=5):
        super().__init__(
            llm=llm,
            system_prompt=MATH_AGENT_PROMPT,
            tools=MATH_TOOLS,
            max_steps=max_steps,
        )