Building a CrewAI tool for retrieval-augmented generation

This crewai rag tool tutorial shows how to build a custom retrieval tool that gives a CrewAI agent grounded answers from your own documents. We’ll use Chroma for vector storage and an OpenAI-compatible embedding model, then wrap the lookup in a CrewAI BaseTool so the agent can call it autonomously.

Step 1: Install dependencies and configure API access

Start with a clean virtual environment and install the libraries we need:

pip install crewai chromadb openai tiktoken python-dotenv

CrewAI expects an OpenAI-compatible LLM by default. Export your API key and base URL (we’ll cover swapping the base URL later):

export OPENAI_API_KEY=sk-your-key
# optional, defaults to OpenAI
export OPENAI_BASE_URL=https://api.openai.com/v1

Opinion: don’t hardcode keys. Load them from .env with python-dotenv and keep your ingestion script separate from runtime agent code. Ingestion is a batch job; the tool is a latency-sensitive path.

Step 2: Chunk and ingest documents into Chroma

Retrieval quality lives or dies by chunking. A naive fixed-size split is fine for a first cut, but you should add overlap and respect section boundaries in real code. Below is a minimal ingestion script that embeds a markdown spec and stores it in an in-memory Chroma collection.

import chromadb
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client = OpenAI()
chroma = chromadb.Client()
col = chroma.create_collection("docs")

def embed(texts):
    resp = client.embeddings.create(model="text-embedding-3-small", input=texts)
    return [d.embedding for d in resp.data]

with open("spec.md") as f:
    text = f.read()

# 500-char chunks, 50-char overlap
step = 500
overlap = 50
chunks = []
for i in range(0, len(text), step - overlap):
    chunks.append(text[i:i+step])

embeds = embed(chunks)
col.add(
    documents=chunks,
    embeddings=embeds,
    ids=[f"c{i}" for i in range(len(chunks))]
)
print(f"Ingested {len(chunks)} chunks")

Run it once. If you change the source doc, delete the collection or version the collection name (spec_v2). Chroma’s query later uses cosine similarity by default, which is correct for text-embedding-3-small.

A note on scale: in-memory Chroma is fine for a few thousand chunks. Past that, run Chroma as a server or move to a managed vector DB. The tool code below stays identical because it only calls collection.query.

Step 3: Define the RAG tool as a CrewAI custom tool

CrewAI tools are subclasses of BaseTool with a Pydantic args_schema. The description field is not decoration—the agent reads it to decide whether to call your tool. Be specific about when it should be used.

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
import chromadb
from openai import OpenAI

client = OpenAI()
chroma = chromadb.Client()
col = chroma.get_collection("docs")

def _embed(query):
    return client.embeddings.create(
        model="text-embedding-3-small", input=[query]
    ).data[0].embedding

class RetrieverInput(BaseModel):
    query: str = Field(description="A specific question or keyword to search the docs")

class RetrieverTool(BaseTool):
    name: str = "document_retriever"
        "Retrieve relevant passages from the internal product spec. "
        "Use this before answering any question about API behavior, limits, or fields."
    )
    args_schema: type[BaseModel] = RetrieverInput

    def _run(self, query: str) -> str:
        q_emb = _embed(query)
        res = col.query(query_embeddings=[q_emb], n_results=3)
        docs = res["documents"][0]
        return "\n---\n".join(docs)

Key details:

• n_results=3 keeps the context window small. Returning ten chunks wastes tokens and confuses the agent.
• Joining with \n---\n gives the LLM a visible separator between sources.
• The tool returns raw text. Let the agent synthesize; don’t try to format a final answer inside the tool.

If you prefer the decorator style, CrewAI also supports @tool, but a class gives you schema validation and cleaner reuse across agents.

Step 4: Wire the tool into an agent and a crew

Now we give the tool to an agent and run a task. Set verbose=True so you can see the tool call in logs.

from crewai import Agent, Crew, Task

retriever = RetrieverTool()

researcher = Agent(
    role="Documentation specialist",
    goal="Answer questions using only retrieved context",
    backstory="You never guess. You search the spec first, then quote.",
    tools=[retriever],
    verbose=True,
    allow_delegation=False
)

task = Task(
    description="What does the API return when a request is rate limited?",
    expected_output="A concise answer with the relevant quoted passage from the spec.",
    agent=researcher
)

crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff()
print("FINAL:", result)

The agent will reason: “I need to know rate limit behavior → call document_retriever → get passages → answer.” If your description is vague, the agent may skip the tool and hallucinate. That’s the most common failure mode in any crewai rag tool tutorial, so tune the description against real queries.

Step 5: Execute and verify retrieval works

Run the crew script. In the verbose log you should see a line similar to:

Agent downloaded action: document_retriever with args {'query': 'rate limit response'}

Success criteria:

1. The log shows the tool was called (not skipped).
2. The final printed answer contains text that exists in spec.md.
3. If you delete the Chroma collection, the answer degrades or the agent says it can’t find info.

Add a fast unit test for the tool alone so you don’t need the full crew to debug retrieval:

def test_retriever():
    t = RetrieverTool()
    out = t.run("rate limit")
    assert len(out) > 20
    # optional: assert a known phrase from spec.md appears
    assert "429" in out or "Too Many Requests" in out

Run pytest after every chunking change. Retrieval regression is silent until the agent starts misanswering.

Step 6: Swap to a unified OpenAI-compatible gateway

If you want one endpoint for both embeddings and chat completions, point the OpenAI client at n4n.ai’s OpenAI-compatible endpoint that addresses 240+ models, with automatic fallback when a provider is rate-limited or degraded. It honors client routing directives and forwards provider cache-control hints. Change only the environment:

export OPENAI_BASE_URL=https://api.n4n.ai/v1

No code edits are required because the tool and CrewAI both use the OpenAI SDK under the hood. This removes the need to juggle multiple provider keys when you later upgrade the agent to a different model—just change the model string.

Production hardening tips

The code above is a complete crewai rag tool tutorial, but before shipping, address three things:

• Metadata filtering: add a metadata dict per chunk (e.g., section: "auth") and pass where= to col.query so the agent can narrow by domain.
• Hybrid search: Chroma supports keyword + vector. Expose a mode param in RetrieverInput if the agent needs exact-string lookup.
• Cache embeddings: never re-embed the same query twice in a single run; the OpenAI client won’t cache for you. A 60-second TTL dict is enough.

Build the tool, test it standalone, then let the agent loose. Retrieval-augmented generation stops being magic the moment you can see the chunks the model actually received.