feat(rag): add pure Python chunking and RRF foundation

apps/rag-pipeline/.gitignore

@@ -0,0 +1,3 @@
+__pycache__/
+*.py[cod]
+.venv/

apps/rag-pipeline/package.json

@@ -0,0 +1,8 @@
+{
+  "name": "@agentx/rag-pipeline",
+  "version": "0.1.0",
+  "private": true,
+  "scripts": {
+    "test": "python3 -m unittest discover -s tests -v"
+  }
+}

apps/rag-pipeline/pyproject.toml

@@ -0,0 +1,13 @@
+[project]
+name = "agentx-rag-core"
+version = "0.1.0"
+description = "Pure Python text chunking and rank-fusion foundations for AgentX RAG"
+requires-python = ">=3.11"
+dependencies = []
+
+[build-system]
+requires = ["hatchling==1.27.0"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["rag"]

apps/rag-pipeline/rag/__init__.py

@@ -0,0 +1,5 @@
+"""Pure Python RAG foundation utilities."""
+
+from .core import reciprocal_rank_fusion, split_text
+
+__all__ = ["reciprocal_rank_fusion", "split_text"]

apps/rag-pipeline/rag/core.py

@@ -0,0 +1,90 @@
+"""Dependency-free building blocks for retrieval-augmented generation."""
+
+from collections.abc import Hashable, Iterable, Sequence
+from typing import TypeVar
+
+T = TypeVar("T", bound=Hashable)
+
+
+def split_text(
+    text: str,
+    chunk_size: int = 1200,
+    chunk_overlap: int = 300,
+    min_chunk_chars: int = 40,
+) -> list[str]:
+    """Split text into deterministic overlapping character windows.
+
+    When possible, a window ends at the last sentence or newline boundary
+    inside its overlap-sized tail. Leading and trailing whitespace is removed
+    from each returned chunk.
+    """
+    if chunk_size <= 0:
+        raise ValueError("chunk_size must be positive")
+    if chunk_overlap < 0 or chunk_overlap >= chunk_size:
+        raise ValueError("chunk_overlap must be non-negative and smaller than chunk_size")
+    if min_chunk_chars < 0:
+        raise ValueError("min_chunk_chars must be non-negative")
+
+    chunks: list[str] = []
+    start = 0
+    text_length = len(text)
+
+    while start < text_length:
+        end = min(start + chunk_size, text_length)
+        if end < text_length:
+            boundary_floor = max(start, end - chunk_overlap)
+            for index in range(end, boundary_floor, -1):
+                if text[index - 1] in ".!?\n":
+                    end = index
+                    break
+
+        chunk = text[start:end].strip()
+        if len(chunk) >= min_chunk_chars and chunk:
+            chunks.append(chunk)
+
+        if end == text_length:
+            break
+        start = end - chunk_overlap
+
+    return chunks
+
+
+def reciprocal_rank_fusion(
+    rankings: Sequence[Iterable[T]],
+    *,
+    weights: Sequence[float] | None = None,
+    k: float = 60,
+    limit: int | None = None,
+) -> list[tuple[T, float]]:
+    """Fuse ranked item lists using weighted reciprocal rank fusion.
+
+    An item contributes ``weight / (k + rank)`` for each ranking containing
+    it, where ranks are one-based. Duplicate items in a ranking are ignored.
+    Score ties retain the order in which items were first encountered.
+    """
+    if k < 0:
+        raise ValueError("k must be non-negative")
+    if limit is not None and limit < 0:
+        raise ValueError("limit must be non-negative")
+
+    ranking_weights = list(weights) if weights is not None else [1.0] * len(rankings)
+    if len(ranking_weights) != len(rankings):
+        raise ValueError("weights must contain one value per ranking")
+    if any(weight < 0 for weight in ranking_weights):
+        raise ValueError("weights must be non-negative")
+
+    scores: dict[T, float] = {}
+    first_seen: dict[T, int] = {}
+    for ranking, weight in zip(rankings, ranking_weights, strict=True):
+        seen: set[T] = set()
+        rank = 0
+        for item in ranking:
+            if item in seen:
+                continue
+            seen.add(item)
+            rank += 1
+            first_seen.setdefault(item, len(first_seen))
+            scores[item] = scores.get(item, 0.0) + weight / (k + rank)
+
+    fused = sorted(scores.items(), key=lambda pair: (-pair[1], first_seen[pair[0]]))
+    return fused if limit is None else fused[:limit]

apps/rag-pipeline/tests/test_core.py

@@ -0,0 +1,81 @@
+import unittest
+
+from rag import reciprocal_rank_fusion, split_text
+
+
+class SplitTextTests(unittest.TestCase):
+    def test_returns_short_document_when_minimum_is_met(self) -> None:
+        self.assertEqual(split_text("A complete sentence.", min_chunk_chars=1), ["A complete sentence."])
+
+    def test_prefers_sentence_boundary_and_preserves_overlap(self) -> None:
+        text = "Alpha bravo. Charlie delta echo. Foxtrot golf hotel."
+
+        chunks = split_text(text, chunk_size=36, chunk_overlap=8, min_chunk_chars=1)
+
+        self.assertEqual(chunks, ["Alpha bravo. Charlie delta echo.", "ta echo. Foxtrot golf hotel."])
+
+    def test_falls_back_to_fixed_width_when_no_boundary_exists(self) -> None:
+        self.assertEqual(
+            split_text("abcdefghijkl", chunk_size=5, chunk_overlap=2, min_chunk_chars=1),
+            ["abcde", "defgh", "ghijk", "jkl"],
+        )
+
+    def test_skips_blank_and_tiny_chunks(self) -> None:
+        self.assertEqual(split_text("   ", min_chunk_chars=1), [])
+        self.assertEqual(split_text("tiny", min_chunk_chars=5), [])
+
+    def test_rejects_invalid_configuration(self) -> None:
+        for kwargs in (
+            {"chunk_size": 0},
+            {"chunk_overlap": -1},
+            {"chunk_size": 10, "chunk_overlap": 10},
+            {"min_chunk_chars": -1},
+        ):
+            with self.subTest(kwargs=kwargs), self.assertRaises(ValueError):
+                split_text("text", **kwargs)
+
+
+class ReciprocalRankFusionTests(unittest.TestCase):
+    def test_combines_rankings_with_standard_rrf_score(self) -> None:
+        fused = reciprocal_rank_fusion([["a", "b"], ["b", "c"]], k=60)
+
+        self.assertEqual([item for item, _ in fused], ["b", "a", "c"])
+        self.assertAlmostEqual(fused[0][1], 1 / 62 + 1 / 61)
+
+    def test_applies_weights_and_limit(self) -> None:
+        fused = reciprocal_rank_fusion(
+            [["vector-first", "shared"], ["shared", "keyword-first"]],
+            weights=[0.8, 0.2],
+            k=0,
+            limit=2,
+        )
+
+        self.assertEqual([item for item, _ in fused], ["vector-first", "shared"])
+        self.assertAlmostEqual(fused[0][1], 0.8)
+        self.assertAlmostEqual(fused[1][1], 0.6)
+
+    def test_ignores_duplicate_item_within_one_ranking(self) -> None:
+        fused = reciprocal_rank_fusion([["a", "a", "b"]], k=0)
+
+        self.assertEqual(fused, [("a", 1.0), ("b", 0.5)])
+
+    def test_ties_keep_first_seen_order(self) -> None:
+        self.assertEqual(
+            reciprocal_rank_fusion([["a"], ["b"]], k=0),
+            [("a", 1.0), ("b", 1.0)],
+        )
+
+    def test_rejects_invalid_configuration(self) -> None:
+        invalid_calls = (
+            lambda: reciprocal_rank_fusion([["a"]], k=-1),
+            lambda: reciprocal_rank_fusion([["a"]], weights=[1, 2]),
+            lambda: reciprocal_rank_fusion([["a"]], weights=[-1]),
+            lambda: reciprocal_rank_fusion([["a"]], limit=-1),
+        )
+        for call in invalid_calls:
+            with self.subTest(call=call), self.assertRaises(ValueError):
+                call()
+
+
+if __name__ == "__main__":
+    unittest.main()