diff --git a/apps/rag-pipeline/.gitignore b/apps/rag-pipeline/.gitignore new file mode 100644 index 0000000..98dbeb4 --- /dev/null +++ b/apps/rag-pipeline/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +*.py[cod] +.venv/ diff --git a/apps/rag-pipeline/package.json b/apps/rag-pipeline/package.json new file mode 100644 index 0000000..5355cf8 --- /dev/null +++ b/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" + } +} diff --git a/apps/rag-pipeline/pyproject.toml b/apps/rag-pipeline/pyproject.toml new file mode 100644 index 0000000..551b4dd --- /dev/null +++ b/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"] diff --git a/apps/rag-pipeline/rag/__init__.py b/apps/rag-pipeline/rag/__init__.py new file mode 100644 index 0000000..bc22615 --- /dev/null +++ b/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"] diff --git a/apps/rag-pipeline/rag/core.py b/apps/rag-pipeline/rag/core.py new file mode 100644 index 0000000..6d976f9 --- /dev/null +++ b/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] diff --git a/apps/rag-pipeline/tests/test_core.py b/apps/rag-pipeline/tests/test_core.py new file mode 100644 index 0000000..3b60613 --- /dev/null +++ b/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() diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 2ada516..f00db99 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -415,6 +415,8 @@ importers: specifier: ^1.3.0 version: 1.3.0 + apps/rag-pipeline: {} + apps/simon-cli: dependencies: '@agentx/core':