Metadata-Version: 2.4
Name: agentic-rag-toolkit
Version: 0.1.0
Summary: Provider-agnostic, drop-in agent primitives for agentic RAG pipelines: classification, retrieval routing, sub-question retrieval, quality gating, self-validation, and tone personalization.
Project-URL: Homepage, https://github.com/mohanapriya-sk/agentic-rag-toolkit
Project-URL: Repository, https://github.com/mohanapriya-sk/agentic-rag-toolkit
Project-URL: Issues, https://github.com/mohanapriya-sk/agentic-rag-toolkit/issues
Author-email: Mohanapriya S <mohanapriyas.mca@gmail.com>
License: MIT
License-File: LICENSE
Keywords: agentic-ai,langchain,llamaindex,llm,quality-gate,rag,retrieval-augmented-generation
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.9
Requires-Dist: pydantic>=2.0
Provides-Extra: chromadb
Requires-Dist: chromadb>=0.4; extra == 'chromadb'
Provides-Extra: dev
Requires-Dist: build; extra == 'dev'
Requires-Dist: pytest-cov; extra == 'dev'
Requires-Dist: pytest>=7.0; extra == 'dev'
Requires-Dist: ruff; extra == 'dev'
Requires-Dist: twine; extra == 'dev'
Provides-Extra: ollama
Requires-Dist: ollama>=0.1; extra == 'ollama'
Provides-Extra: openai
Requires-Dist: openai>=1.0; extra == 'openai'
Provides-Extra: pinecone
Requires-Dist: pinecone-client>=3.0; extra == 'pinecone'
Description-Content-Type: text/markdown

# agentic-rag-toolkit

Drop-in agent primitives for pipelines you already have -- not another framework.

`agentic-rag-toolkit` is a small Python library of **six reusable agent building blocks** distilled from production agentic-RAG work: a clinical/behavioral-health nudge engine and an LLM-powered lead-generation pipeline. Each block does one job -- classify, select a retriever, retrieve, quality-check, self-validate, or personalize tone -- and any of them can be dropped into an existing LangChain/LlamaIndex/raw-Python pipeline without adopting a whole new framework. Works with any LLM backend (Ollama, OpenAI/GPT-4o) and any vector store (Chroma, Pinecone, or your own) through one shared interface.

## Why this exists

| Existing tool | What it actually does | Why it's not the same |
|---|---|---|
| `raki`, `rag-evaluation` | Score RAG quality **after** a session is over (post-hoc analysis) | Never runs *inside* the live pipeline -- can't stop a bad answer before the user sees it |
| `create-agentic-rag` | A project scaffold/boilerplate generator | One-time template, not an importable library you keep using |
| `flashrag-dev`, `mini-rag` | Full end-to-end batteries-included RAG research pipelines | You adopt their entire chunk-embed-retrieve-generate flow -- can't cherry-pick one piece |
| LangGraph / LlamaIndex ADW | Quality gates and self-correction exist as internal concepts | Only usable if you buy into their full state-machine/orchestration model |
| `pydantic-ai-rag` | RAG layer bolted onto one specific agent framework | Locked to pydantic-ai, not portable |

Three things to lead with:

1. **Runtime gating, not post-hoc scoring.** `QualityGateAgent` blocks a bad answer *before* the user sees it.
2. **Composable, not all-or-nothing.** Import just `QualityGateAgent` into an existing LangChain chain -- no rewrite required.
3. **Provider-agnostic by contract.** One `LLMClient` interface, swap Ollama <-> OpenAI with one line. One `Retriever` interface, swap Chroma <-> Pinecone with one line.

## Install

```bash
pip install agentic-rag-toolkit          # core only
pip install "agentic-rag-toolkit[ollama]"    # + local Ollama support
pip install "agentic-rag-toolkit[openai]"    # + OpenAI/GPT-4o support
pip install "agentic-rag-toolkit[chromadb]"  # + Chroma example support
```

## Quickstart

```python
from agentic_rag_toolkit import MetricClassifierAgent, QualityGateAgent

class MyLLM:
    def complete(self, prompt: str, **kwargs) -> str:
        ...  # call your model here, return raw text

llm = MyLLM()

classifier = MetricClassifierAgent(llm=llm, categories=["urgent", "routine"])
result = classifier.run("heart rate spiked to 180")
print(result.label, result.confidence)

gate = QualityGateAgent(llm=llm, min_evidence_count=1)
gate_result = gate.run(evidence=["some retrieved doc"], query="what happened?")
if not gate_result.passed:
    print("blocked:", gate_result.reason)
```

See `examples/basic_pipeline.py` for a full runnable demo (no API keys needed) chaining all six agents, and `examples/ollama_chroma_pipeline.py` for a production-shaped version with Ollama + Chroma.

## API Reference

All LLM-facing agents ask for structured JSON output (validated with Pydantic, with automatic retry-with-error-feedback on a bad response) instead of parsing ad-hoc text formats like `"label|0.92"` -- see `complete_json` below. Every agent exposes a single `.run(...)` entrypoint.

### `LLMClient` (Protocol) -- `agentic_rag_toolkit.core.llm_client`

Any object with a matching `.complete()` method satisfies this automatically (structural typing via `@runtime_checkable` -- no subclassing required).

```python
class LLMClient(Protocol):
    def complete(self, prompt: str, **kwargs) -> str: ...
```

Two ready-made implementations ship with the toolkit:

| Class | Constructor args | Requires |
|---|---|---|
| `OllamaClient` | `model: str = "llama3"` | `pip install ollama` + Ollama daemon running locally |
| `OpenAIClient` | `model: str = "gpt-4o"` | `pip install openai` + `OPENAI_API_KEY` env var set |

```python
from agentic_rag_toolkit import OllamaClient, OpenAIClient

llm = OllamaClient(model="llama3.1")   # or:
llm = OpenAIClient(model="gpt-4o")
llm.complete("Say hello in one word.")   # -> "Hello"
```

To use your own backend (Anthropic, a local vLLM server, etc.), just implement `.complete(prompt: str, **kwargs) -> str` on any class -- every agent below accepts it.

---

### `complete_json()` -- `agentic_rag_toolkit.core.structured`

The shared helper every LLM-facing agent uses internally. Public and importable if you want the same schema-validated-JSON-with-retry behavior in your own custom agents.

```python
def complete_json(
    llm: LLMClient,
    prompt: str,
    schema: type[BaseModel],
    max_retries: int = 2,
) -> BaseModel
```

| Argument | Type | Description |
|---|---|---|
| `llm` | `LLMClient` | Any object with `.complete()` |
| `prompt` | `str` | Your instruction text (the JSON schema is appended automatically) |
| `schema` | `type[BaseModel]` | A Pydantic model class describing the expected response shape |
| `max_retries` | `int`, default `2` | Extra attempts if the response fails to parse/validate. Total attempts = `max_retries + 1` |

**Returns:** a validated instance of `schema`.
**Raises:** `ValueError` if every attempt fails (message includes the last raw response and validation error).

```python
from pydantic import BaseModel
from agentic_rag_toolkit import complete_json

class Sentiment(BaseModel):
    positive: bool
    reason: str

result = complete_json(llm, "Is this review positive? 'Loved it!'", Sentiment)
print(result.positive, result.reason)
```

---

### `MetricClassifierAgent` -- `agentic_rag_toolkit.classifiers.metric_classifier`

Labels incoming text into one of a fixed set of categories -- the first decision point in a pipeline.

**Constructor**

```python
MetricClassifierAgent(llm: LLMClient, categories: list[str], max_retries: int = 2)
```

| Argument | Type | Default | Notes |
|---|---|---|---|
| `llm` | `LLMClient` | required | |
| `categories` | `list[str]` | required | Must be non-empty -- raises `ValueError` otherwise |
| `max_retries` | `int` | `2` | Passed through to `complete_json` |

**Method**

```python
.run(input_text: str) -> ClassificationResult
```
`ClassificationResult` is `{label: str, confidence: float}` (confidence constrained to `0.0-1.0`).

Raises `ValueError` if `input_text` is empty/blank, if the LLM returns a label outside `categories`, or if the response can't be parsed after retries.

```python
from agentic_rag_toolkit import MetricClassifierAgent

classifier = MetricClassifierAgent(llm=llm, categories=["urgent", "routine"])
result = classifier.run("heart rate spiked to 180")
# result.label == "urgent", result.confidence == 0.91
```

---

### `RetrieverSelectorAgent` + `Retriever` -- `agentic_rag_toolkit.retrieval.retriever_selector`

Routes a classified category to the correct retriever. Pure lookup -- **does not call an LLM**, so it takes no `llm` argument (unlike every other agent here).

`Retriever` is a Protocol: any object with `.retrieve(query: str, top_k: int = 5) -> list[str]` qualifies -- wrap your Chroma collection, Pinecone index, or anything else this way.

**Constructor**

```python
RetrieverSelectorAgent(retriever_map: dict[str, Retriever])
```
`retriever_map` must be non-empty -- raises `ValueError` otherwise.

**Method**

```python
.run(category: str) -> Retriever
```
Raises `KeyError` (with the list of known categories in the message) if `category` isn't in `retriever_map`.

```python
from agentic_rag_toolkit import RetrieverSelectorAgent

selector = RetrieverSelectorAgent(retriever_map={
    "urgent": urgent_chroma_wrapper,
    "routine": routine_chroma_wrapper,
})
retriever = selector.run("urgent")
docs = retriever.retrieve("blood sugar spike guidance")
```

---

### `SubQuestionRetrieverAgent` -- `agentic_rag_toolkit.retrieval.sub_question_retriever`

Breaks one complex question into smaller sub-questions and retrieves evidence for each **in parallel**, rather than one broad fuzzy search.

**Constructor**

```python
SubQuestionRetrieverAgent(
    llm: LLMClient,
    retriever: Retriever,
    max_sub_questions: int = 4,
    max_retries: int = 2,
)
```

| Argument | Type | Default | Notes |
|---|---|---|---|
| `llm` | `LLMClient` | required | Used only to decompose the question |
| `retriever` | `Retriever` | required | Any object with `.retrieve()` |
| `max_sub_questions` | `int` | `4` | Caps how many sub-questions are generated (and therefore worker threads spawned) |
| `max_retries` | `int` | `2` | Retries for the decomposition call |

**Method**

```python
.run(question: str) -> dict[str, list[str]]
```
Returns a dict mapping each generated sub-question to its retrieved evidence list. Raises `ValueError` if `question` is empty. If decomposition fails validation even after retries, it silently falls back to treating the original question as the only sub-question -- decomposition is a nice-to-have, not a hard requirement.

```python
from agentic_rag_toolkit import SubQuestionRetrieverAgent

agent = SubQuestionRetrieverAgent(llm=llm, retriever=my_retriever, max_sub_questions=3)
results = agent.run("What should happen when a patient's blood sugar spikes?")
# {"What is a safe blood sugar range?": [...], "What should someone do right now?": [...]}
```

---

### `QualityGateAgent` + `GateResult` -- `agentic_rag_toolkit.quality.quality_gate`

**The core differentiator of this toolkit.** Checks retrieved evidence is sufficient *before* generation happens, so the pipeline can stop instead of letting the LLM hallucinate.

**Constructor**

```python
QualityGateAgent(llm: LLMClient, min_evidence_count: int = 1, max_retries: int = 2)
```

| Argument | Type | Default | Notes |
|---|---|---|---|
| `llm` | `LLMClient` | required | |
| `min_evidence_count` | `int` | `1` | If fewer evidence items are passed in, the gate fails immediately -- no LLM call needed |
| `max_retries` | `int` | `2` | |

**Method**

```python
.run(evidence: list[str], query: str) -> GateResult
```
`GateResult` is `{passed: bool, reason: str}`. Raises `ValueError` if the LLM's sufficiency judgment can't be parsed after retries.

```python
from agentic_rag_toolkit import QualityGateAgent

gate = QualityGateAgent(llm=llm, min_evidence_count=2)
result = gate.run(evidence=["doc A", "doc B"], query="what is a safe blood sugar range?")
if not result.passed:
    print("blocked:", result.reason)
```

---

### `SelfValidatorAgent` + `ValidationResult` -- `agentic_rag_toolkit.quality.self_validator`

Final compliance/safety check on a *generated* answer -- runs after your own generation step, against rules you define.

**Constructor**

```python
SelfValidatorAgent(llm: LLMClient, rules: list[str], max_retries: int = 2)
```
`rules` must be non-empty -- raises `ValueError` otherwise.

**Method**

```python
.run(generated_answer: str) -> ValidationResult
```
`ValidationResult` is `{valid: bool, violated_rules: list[str]}`. Raises `ValueError` if `generated_answer` is empty.

```python
from agentic_rag_toolkit import SelfValidatorAgent

validator = SelfValidatorAgent(llm=llm, rules=[
    "never state a specific medication dosage",
    "always suggest consulting a doctor if severe",
])
result = validator.run("Take 500mg twice a day.")
# result.valid == False, "never state a specific medication dosage" in result.violated_rules
```

---

### `TonePersonalizerAgent` -- `agentic_rag_toolkit.personalization.tone_personalizer`

Rewrites an already-approved answer into a target voice, changing delivery only -- never facts. Meant to run **last**, after `QualityGateAgent` and `SelfValidatorAgent` have both passed, so a nicer tone can never mask a bad or non-compliant answer. Returns plain text, not structured JSON -- there's no schema for "does this sound gentle enough."

**Constructor**

```python
TonePersonalizerAgent(llm: LLMClient, tone: str)
```
`tone` must be non-empty -- raises `ValueError` otherwise.

**Method**

```python
.run(approved_answer: str) -> str
```
Raises `ValueError` if `approved_answer` is empty.

```python
from agentic_rag_toolkit import TonePersonalizerAgent

personalizer = TonePersonalizerAgent(llm=llm, tone="gentle and encouraging, like a supportive friend")
message = personalizer.run("Your blood sugar reading is high.")
```

---

Full end-to-end wiring of all six agents: `examples/basic_pipeline.py` (no API keys needed) and `examples/ollama_chroma_pipeline.py` (production-shaped, Ollama + Chroma).

## Development

```bash
python -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install -e ".[dev]"
pytest --cov=agentic_rag_toolkit tests/ -v
```

Release process for maintainers lives in `PUBLISHING.md`, not here -- this README is for people using the package, not publishing it.

## License

MIT -- see `LICENSE`.