Most RAG demos break because they treat retrieval as a single fixed step before the model speaks. To build agentic RAG pipeline that routes queries, rewrites them, and calls tools inside a reasoning loop, you need the LLM to control retrieval calls. This tutorial implements a minimal but production-shaped version in Python.
Prerequisites
- Python 3.10+ and
pip install openai chromadb. - An OpenAI-compatible endpoint. If you want one key for 240+ models with automatic fallback when a provider is degraded, point the SDK at n4n.ai’s gateway.
- A folder
./docswith a few.txtfiles to index.
We use Chroma’s built-in embedding model to avoid extra services. Swap in your own embeddings later.
1. Ingest documents into a vector store
Create a local Chroma collection and load text files. Keep chunking dumb simple: split on double newlines.
import chromadb, os, glob
client = chromadb.PersistentClient(path="./chroma")
coll = client.get_or_create_collection("docs")
paths = glob.glob("./docs/*.txt")
for i, p in enumerate(paths):
text = open(p).read()
chunks = [c for c in text.split("\n\n") if c.strip()]
coll.add(
ids=[f"{i}-{j}" for j in range(len(chunks))],
documents=chunks,
metadatas=[{"source": p}] * len(chunks),
)
print("indexed", coll.count(), "chunks")Expected output after running:
indexed 42 chunks2. Define the retrieval tool
The agent calls this via function calling. Return concatenated matches with source tags so the model can cite.
def retrieve(query: str, top_k: int = 3) -> str:
res = coll.query(query_texts=[query], n_results=top_k)
docs = res["documents"][0]
metas = res["metadatas"][0]
out = []
for d, m in zip(docs, metas):
out.append(f"[{m['source']}] {d}")
return "\n---\n".join(out)Test it directly:
print(retrieve("what is the refund policy?"))Expected shape:
[./docs/terms.txt] Refunds are issued within 30 days of purchase...
---
[./docs/faq.txt] If your item is defective, email support@...3. Build the agent loop
Define the tool schema and run a loop: send messages, if the model emits tool_calls, execute them and feed results back. Stop when the model replies with text.
from openai import OpenAI
import json
client = OpenAI(base_url="https://api.n4n.ai/v1", api_key=os.environ["API_KEY"])
tools = [{
"type": "function",
"function": {
"name": "retrieve",
"description": "Search internal docs for a query",
"parameters": {
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
},
},
}]
system = "You are a research agent. Use retrieve to answer from internal docs. Cite sources."
def run_agent(question: str, max_steps: int = 5):
msgs = [{"role": "system", "content": system},
{"role": "user", "content": question}]
for _ in range(max_steps):
resp = client.chat.completions.create(
model="gpt-4o-mini", messages=msgs, tools=tools)
msg = resp.choices[0].message
if not msg.tool_calls:
return msg.content
msgs.append(msg)
for tc in msg.tool_calls:
args = json.loads(tc.function.arguments)
result = retrieve(args["query"])
msgs.append({"role": "tool", "tool_call_id": tc.id,
"content": result})
return "Agent exceeded step budget"This is the core of how you build agentic RAG pipeline: the model decides when to fetch, not your glue code.
4. Add routing and query rewriting
A single retrieve call is rarely enough. Agentic behavior emerges when the model issues multiple calls with rephrased queries. Force that by tightening the system prompt and adding a second stub tool for external lookup.
tools.append({
"type": "function",
"function": {
"name": "web_search",
"description": "Search the public web",
"parameters": {
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
},
},
})
def web_search(query: str) -> str:
# stub: replace with real API
return f"Web results for '{query}' (not implemented)"Now the agent can route between internal and external knowledge. In practice, the model will call retrieve with a decomposed sub-question, then web_search if the docs are silent.
To make rewriting explicit, log the arguments:
for tc in msg.tool_calls:
print("TOOL:", tc.function.name, tc.function.arguments)You’ll see output like:
TOOL: retrieve {"query": "refund timeframe for defective items"}
TOOL: web_search {"query": "EU consumer protection refund 2024"}That transcript is your evidence the pipeline is agentic, not a linear retriever-generator.
5. Run end-to-end
Wire a CLI entry point and inspect the full exchange.
if __name__ == "__main__":
ans = run_agent("How do I get a refund for a broken widget bought last week?")
print("ANSWER:", ans)Sample answer (truncated):
ANSWER: According to ./docs/terms.txt, refunds are issued within 30 days of purchase.
Since you bought it last week, email support@acme.com with your order number [source: faq.txt].If the agent hit the web tool, you’d see combined citations. The loop terminated because the final message had no tool_calls.
6. Production hardening
Set max_steps to bound cost. Stream responses to avoid blocking the user: pass stream=True and iterate resp. Use a gateway that honors client routing directives and forwards provider cache-control hints—n4n.ai does this, so your retrieval prompt prefix can be cached by the upstream provider when you send cache_control markers.
Per-token usage metering matters when the agent self-loops. Capture resp.usage on each call and sum it:
usage = resp.usage
total_tokens += usage.total_tokensAdd a retry with exponential backoff around client.chat.completions.create. When a provider is rate-limited, an inference gateway with automatic fallback switches models silently; your code just sees a successful completion.
Finally, persist msgs to reconstruct sessions. Agentic RAG is stateful—the second question benefits from the first retrieval.
Where to take it further
Replace Chroma with a managed vector DB when docs exceed a few GB. Swap gpt-4o-mini for a cheaper model on simple rewrites and a stronger one on synthesis. The loop you built is the same primitive used in larger orchestration frameworks; you now know exactly what they hide.
Keep the tool contracts strict, log every tool_call, and you can build agentic RAG pipeline that survives contact with real documents.