Metadata-Version: 2.4
Name: right-rag
Version: 0.1.0
Summary: Domain-structured RAG pipeline template
License-File: LICENSE
Requires-Python: >=3.13
Requires-Dist: docling>=2.110.0
Requires-Dist: prefect>=3.7.7
Requires-Dist: pydantic>=2.13.4
Description-Content-Type: text/markdown

﻿# 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:

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

Or with uv:

```powershell
uvx right-rag init my-rag
cd my-rag
```

Add source documents:

```text
data/documents/
```

Run the pipeline:

```powershell
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:

```powershell
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:

```powershell
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:

```python
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:

```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:

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

## Project Layout

```text
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:

```python
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:

```python
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:

```text
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:

```json
{
  "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:

```powershell
python -m compileall src/right_rag
```

Run the test suite:

```powershell
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:

```powershell
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:

```powershell
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.
