right-rag 0.1.0

Domain-structured RAG pipeline template

right-rag

Domain-structured RAG pipeline for documents where plain text chunks lose too much context: laws, policies, contracts, manuals, academic papers, handbooks, and other hierarchical corpora.

right-rag keeps the pipeline core generic. You adapt it by editing or adding plugins under spec/ so the pipeline matches your documents.

Install And Start

Requirements:

• Python 3.13+

For normal use, install and create a project:

pip install right-rag
right-rag init my-rag
cd my-rag

Or with uv:

uvx right-rag init my-rag
cd my-rag

Add source documents:

data/documents/

Run the pipeline:

right-rag index
right-rag query "your question"
right-rag evaluate

Generated artifacts:

• units: output/units/
• chunks: output/chunks/
• incremental state: .right-rag-state.json
• parser coverage: stored under each unit parser in .right-rag-state.json

These are local runtime artifacts and are ignored by Git.

For contributing to the framework itself:

git clone https://github.com/sunmodza/right-rag.git right-rag
cd right-rag
uv sync
python -m unittest discover -s tests

Release

CI runs tests and builds the package on push to main and pull requests.

Publishing to PyPI runs when a version tag is pushed:

git tag v0.1.0
git push origin v0.1.0

Configure PyPI Trusted Publishing for this repository before the first release:

• project name: right-rag
• workflow: publish.yml
• environment: pypi

Adapt It To Your Documents

This project is not a plain text chunker. Production quality comes from domain-specific structure: the parser must know what a section, clause, definition, duty, penalty, procedure step, or equivalent domain unit is.

Do not use the default unit parser or default chunk parser with production data. They are smoke-test parsers only, useful for checking that loading, state, saving, and plugin discovery work. For real retrieval, write parsers in spec/ that match your document families.

Start by looking at your source documents, not by editing the pipeline core. Decide what the real domain unit is:

• laws: act -> chapter -> section -> clause -> subclause
• contracts: agreement -> article -> clause -> schedule item
• policies: policy -> section -> rule -> exception
• manuals: manual -> chapter -> procedure -> step
• academic papers: paper -> section -> subsection -> paragraph/table/figure note

Then edit or add plugins:

• spec/document_loaders/: use when the source is not a normal local document folder.
• spec/unit_parsers/: convert each document into a faithful ParsedUnit tree.
• spec/chunk_parsers/: convert each Unit into retrieval-ready Chunk records.
• spec/unit_savers/: change where units are written.
• spec/chunk_savers/: change where chunks are written.

Each plugin should expose a no-argument class that inherits the matching interface. The registry auto-discovers concrete classes under spec/.

Prefer changing spec/ over changing src/right_rag/pipeline.py, src/right_rag/model.py, or src/right_rag/pipeline_state.py.

How to Design Parsers

1. Survey the corpus first. Count how many document formats you really have.
2. Split parsers by document format, not by convenience. One parser can handle one stable family; many formats should have many parsers.
3. Identify structural markers for each format: หมวด, มาตรา, ข้อ, Section, Chapter, definitions, penalty provisions, duties, prohibitions, transitory provisions, tables, schedules, or appendices.
4. Make each UnitParser return a ParsedUnit tree, not raw chunks.
5. Make each ChunkParser chunk from semantic Unit objects after structure exists.

Every unit parser must implement parser_can_handle_document() narrowly. It should accept only the document family it owns. Avoid catch-all parsers in production. If a parser cannot identify the structure, reject/report the reason instead of falling back to one giant unit or arbitrary text chunks.

Example non-overlapping unit parsers:

from __future__ import annotations

import re
from collections.abc import Iterable

from right_rag.interfaces import UnitParser
from right_rag.model import Document, ParsedUnit


class ThaiOrganicActParser(UnitParser):
    @property
    def parser_name(self) -> str:
        return "thai-organic-act"

    def parser_can_handle_document(self, document: Document) -> bool:
        text = document.text[:4000]
        return "พระราชบัญญัติประกอบรัฐธรรมนูญ" in text and "มาตรา" in text

    def parse_document(self, document: Document) -> Iterable[ParsedUnit]:
        for match in re.finditer(r"(?m)^มาตรา\s+(\d+)\s+(.+?)(?=^มาตรา\s+\d+|\Z)", document.text, re.S):
            yield ParsedUnit(
                topic=f"Section {match.group(1)}",
                fulltext=match.group(0).strip(),
                source_start=match.start(),
                source_end=match.end(),
                attributes={"unit_type": "section", "role": "rule"},
            )


class ThaiOrderParser(UnitParser):
    @property
    def parser_name(self) -> str:
        return "thai-order"

    def parser_can_handle_document(self, document: Document) -> bool:
        text = document.text[:4000]
        return "คำสั่ง" in text and "ข้อ" in text and "พระราชบัญญัติประกอบรัฐธรรมนูญ" not in text

    def parse_document(self, document: Document) -> Iterable[ParsedUnit]:
        for match in re.finditer(r"(?m)^ข้อ\s+(\d+)\s+(.+?)(?=^ข้อ\s+\d+|\Z)", document.text, re.S):
            yield ParsedUnit(
                topic=f"Clause {match.group(1)}",
                fulltext=match.group(0).strip(),
                source_start=match.start(),
                source_end=match.end(),
                attributes={"unit_type": "clause", "role": "order"},
            )


class EnglishActParser(UnitParser):
    @property
    def parser_name(self) -> str:
        return "english-act"

    def parser_can_handle_document(self, document: Document) -> bool:
        text = document.text[:4000]
        return bool(re.search(r"\bAct\b", text)) and bool(re.search(r"(?m)^Section\s+\d+", document.text))

    def parse_document(self, document: Document) -> Iterable[ParsedUnit]:
        for match in re.finditer(r"(?m)^Section\s+(\d+)\.?\s+(.+?)(?=^Section\s+\d+|\Z)", document.text, re.S):
            yield ParsedUnit(
                topic=f"Section {match.group(1)}",
                fulltext=match.group(0).strip(),
                source_start=match.start(),
                source_end=match.end(),
                attributes={"unit_type": "section", "role": "rule"},
            )

Usage

By default, the pipeline uses every registered implementation:

• document loaders
• unit parsers
• chunk parsers
• unit savers
• chunk savers

Configure a run from Python:

from right_rag.pipeline import PipelineConfig, pipeline


pipeline(
    PipelineConfig(
        document_loader_names=["default"],
        unit_parser_names=["default"],
        chunk_parser_names=["default"],
        unit_saver_names=["default"],
        chunk_saver_names=["default"],
    )
)

None means "use all registered implementations". An empty list means "use none".

For production, pass explicit parser names and exclude default unit/chunk parsers:

pipeline(
    PipelineConfig(
        unit_parser_names=["thai-organic-act", "thai-order", "english-act"],
        chunk_parser_names=["domain-semantic"],
    )
)

Project Layout

right-rag/
  src/right_rag/
    interfaces.py            Plugin contracts
    model.py                 Document, ParsedUnit, Unit, Chunk, change helpers
    pipeline.py              Pipeline orchestration
    pipeline_state.py        JSON state, stable hashing, serialization
    registry.py              Plugin discovery and factories
    cli.py                   right-rag command line entrypoint
  spec/
    document_loaders/        DocumentLoader plugins
    unit_parsers/            UnitParser plugins
    chunk_parsers/           ChunkParser plugins
    unit_savers/             UnitSaver plugins
    chunk_savers/            ChunkSaver plugins
  tests/                     unittest suite for core and default plugins

Default paths:

• source documents: data/documents/
• unit output: output/units/
• chunk output: output/chunks/
• state: .right-rag-state.json

Built-ins

• document loader: default (Docling)
• unit parser: default (line-based smoke test only)
• chunk parser: default (line-based smoke test only)
• unit saver: default
• chunk saver: default

The default parsers are intentionally dumb. They exist to prove that the pipeline runs; they are not a baseline RAG strategy. Do not include them in production configs unless you are deliberately running a smoke test.

The default document loader reads PDFs with the lightweight pypdfium2 text layer first to avoid Docling StandardPdfPipeline OCR/preprocess memory errors. Other supported file types still go through Docling. Scanned PDFs without a text layer need a custom OCR loader.

Unit Parser Best Practices

Unit parsers return ParsedUnit, not Unit. The pipeline creates Unit objects, ids, parent links, subunit changes, and replayable change history. Source spans and raw text are copied from ParsedUnit onto Unit.

Good unit parsing is the main accuracy layer. It should be faithful to the document, not clever.

• Preserve the author's hierarchy. Do not flatten clauses or sections if the document has meaningful structure.
• Keep topic stable and normalized: Section 12, Clause 4.2, Methods.
• Use temporal_topic only for version/date context the parser can identify: Procurement Act 2560 Section 12.
• Put nested structure in ParsedUnit.subunits.
• Set source_start and source_end to character offsets in Document.text.
• The library computes Unit.raw_text from Document.text[source_start:source_end].
• Put exact source evidence in attributes or metadata: page, heading path, article number, clause number, table id, citation, source filename.
• Keep fulltext faithful. Do not summarize, rewrite, translate, or infer missing content in the unit parser.
• Do not create fake hierarchy because it would be convenient for retrieval.
• Make ids stable by keeping topics stable across reruns.

Minimal example:

from right_rag.interfaces import UnitParser
from right_rag.model import Document, ParsedUnit


class MyUnitParser(UnitParser):
    @property
    def parser_name(self) -> str:
        return "my-parser"

    def parser_can_handle_document(self, document: Document) -> bool:
        return bool(document.text.strip())

    def parse_document(self, document: Document):
        yield ParsedUnit(
            topic="Section 1",
            temporal_topic="Example Act 2024 Section 1",
            fulltext="Normalized unit text",
            source_start=120,
            source_end=260,
            attributes={"section": "1", "page_start": 1, "page_end": 2},
            subunits=[
                ParsedUnit(
                    topic="Subsection 1",
                    fulltext="...",
                    attributes={"subsection": "1"},
                ),
            ],
        )

Chunk Parser Best Practices

Chunk parsers receive normalized Unit objects and return retrieval-friendly Chunk objects.

Good chunking should make answers easy to retrieve and easy to cite.

• Chunk inside a unit; do not mix unrelated sections just to hit a token size.
• Include enough parent context in each chunk: title, section number, temporal topic, and source citation.
• Keep chunks semantically complete. A definition, rule, exception, method, or table row should not be split in the middle if avoidable.
• Use attributes for retrieval filters: document type, jurisdiction, effective date, section number, party, topic, method, dataset, citation.
• Keep fulltext grounded in source text. Add labels or headings only when they improve citation and do not change meaning.
• Avoid embeddings over metadata-only text. Metadata should help filter or cite; source text should carry the answer.
• Prefer fewer high-quality chunks over many tiny fragments.

Minimal example:

from right_rag.interfaces import ChunkParser
from right_rag.model import Chunk, Unit
from right_rag.pipeline_state import stable_hash


class MyChunkParser(ChunkParser):
    @property
    def parser_name(self) -> str:
        return "my-chunker"

    def parse_unit(self, unit: Unit):
        text = f"{unit.temporal_topic}\n\n{unit.fulltext}"
        yield Chunk(
            id=stable_hash({"unit_id": unit.id, "parser": self.parser_name, "index": 1}),
            document_id=unit.document_id,
            unit_id=unit.id,
            topic=unit.topic,
            fulltext=text,
            attributes={
                **unit.attributes,
                "parser_name": self.parser_name,
                "source_unit": unit.temporal_topic,
                "unit_type": unit.attributes.get("unit_type"),
                "hierarchy": unit.attributes.get("hierarchy"),
                "role": unit.attributes.get("role"),
            },
        )

Academically And Practically Accurate RAG

For scholarly, legal, policy, and technical retrieval, accuracy means the answer can be traced back to the source.

Use these rules:

• Keep provenance: source file, page, section, heading path, citation, and table or figure references when available.
• Separate source text from interpretation. Parsers should extract structure; later systems can summarize or reason.
• Preserve version context: date, edition, amendment, jurisdiction, policy version, or publication year.
• Keep boundaries meaningful: a chunk should answer a question without hiding the clause, method, exception, or limitation that controls the answer.
• Record uncertainty in metadata when extraction is heuristic.
• Test with real questions from the domain, not only synthetic examples.
• Inspect output JSON before trusting embeddings.
• Watch parser coverage after each full run. Low coverage means the unit parser is missing parts of the document; suspicious 100% coverage can mean units are too broad.

Anti-Patterns

• One parser accepts every document.
• parser_can_handle_document() returns True for any non-empty text.
• Fallback to ParsedUnit(topic="document", fulltext=document.text).
• Chunking by text length before parsing document structure.
• Using metadata as a substitute for real ParsedUnit.subunits.
• Assuming every document in the corpus has the same format.
• Silently accepting a document when required markers are missing.

Reject unclear input instead. A rejected document with a reason is easier to fix than a confident index full of bad chunks.

Production Checklist

Before a production run, confirm:

• Each parser has a written list of document types it accepts.
• Parser acceptance does not overlap for the same document family.
• Documents with no matching parser are visible in logs/state and reviewed.
• unit_type covers the domain's real units.
• Domain roles are represented: definition, penalty, duty, prohibition, transitory provision, or equivalent roles for your corpus.
• Chunk attributes include parser_name, unit_type, hierarchy, and role.
• No default unit/chunk parser is selected for production.
• No catch-all parser is registered for production.
• No parser falls back to one large unit when structure is missing.

Example expected audit report after a run:

Parser coverage report

documents handled by each parser:
  thai-organic-act: 12
  thai-order: 7
  english-act: 3

documents rejected with reason:
  circular-2024-02.pdf: no Thai order clause markers found
  appendix-a.pdf: appendix format has no registered parser

unit counts by type:
  section: 420
  clause: 180
  definition: 64
  penalty: 18
  transitory_provision: 9

chunks by unit type:
  section: 810
  clause: 300
  definition: 64
  penalty: 36
  transitory_provision: 12

fallback parser used: no

Parser Coverage Development Loop

After uv run right-rag index, inspect .right-rag-state.json. Each unit parser stores coverage like:

{
  "coverage": {
    "document_chars": 10000,
    "covered_chars": 9200,
    "uncovered_chars": 800,
    "coverage_percent": 92.0,
    "units_total": 18,
    "units_with_source_span": 18,
    "source_ranges": [{"start": 0, "end": 9200}]
  }
}

Use it as a development loop:

1. Run the pipeline.
2. Check coverage and saved unit JSON.
3. Improve the unit parser spans or hierarchy.
4. Run again.

Checks

Compile the core files:

python -m compileall src/right_rag

Run the test suite:

python -m unittest discover -s tests

The test suite covers model conversion and change replay, parser coverage, state serialization, registry discovery, and default plugin behavior.

Check plugin discovery:

uv run python -c "from right_rag.registry import build_registry; r=build_registry(); print(r.document_loaders, r.unit_parsers, r.chunk_parsers, r.unit_savers, r.chunk_savers)"

Run the pipeline:

uv run right-rag index

Design Rules

• Put domain logic in spec/.
• Keep src/right_rag/pipeline.py focused on orchestration.
• Return ParsedUnit from unit parsers.
• Let the library create Unit.
• Treat metadata as stored-only context.
• Use stdlib before adding dependencies.