agentic-rag-toolkit

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

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

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

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).

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

Two ready-made implementations ship with the toolkit:

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.

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

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

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

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

Method

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

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

RetrieverSelectorAgent(retriever_map: dict[str, Retriever])

retriever_map must be non-empty -- raises ValueError otherwise.

Method

.run(category: str) -> Retriever

Raises KeyError (with the list of known categories in the message) if category isn't in retriever_map.

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

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

Method

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

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

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

Method

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

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

SelfValidatorAgent(llm: LLMClient, rules: list[str], max_retries: int = 2)

rules must be non-empty -- raises ValueError otherwise.

Method

.run(generated_answer: str) -> ValidationResult

ValidationResult is {valid: bool, violated_rules: list[str]}. Raises ValueError if generated_answer is empty.

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

TonePersonalizerAgent(llm: LLMClient, tone: str)

tone must be non-empty -- raises ValueError otherwise.

Method

.run(approved_answer: str) -> str

Raises ValueError if approved_answer is empty.

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

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.