You read about RAG, vector databases and tool calling and you want to see code running. This post delivers. It is not abstract architecture or a keynote slide, it is a working agent that retrieves context from a knowledge base and answers with grounding, written in Python, with Qdrant and the OpenAI API.
By the end, you have a single file that runs locally, indexes a document, receives a question in the terminal, searches for relevant context via embedding, assembles the prompt, calls the model with tool calling and returns the answer with a cited source. About 120 lines, no framework. The intent is didactic: later you swap it for LangGraph, LlamaIndex or your own framework knowing what is underneath.
1. Stack and installation
Three pieces. The OpenAI SDK for embedding and the LLM, Qdrant on Docker as the vector database, and Python 3.11+. Install the dependencies and bring up Qdrant locally.
pip install openai qdrant-client
# sobe Qdrant em background na porta 6333
docker run -d --name qdrant -p 6333:6333 qdrant/qdrant
# var de ambiente
export OPENAI_API_KEY=sk-...
Create an agent.py file. Structure: setup, ingestion, tool, loop. Let's go step by step.
2. Setup and the Qdrant client
import os
import json
from openai import OpenAI
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
EMBED_MODEL = "text-embedding-3-small"
CHAT_MODEL = "gpt-4o-mini"
COLLECTION = "kb"
oai = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
qdrant = QdrantClient(host="localhost", port=6333)
# cria coleção (1536 = dimensão do text-embedding-3-small)
qdrant.recreate_collection(
collection_name=COLLECTION,
vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
)
3. The ingestion pipeline
Here is what separates a demo from a serious thing: chunking that respects the semantic boundary and enrichment with metadata. To keep it simple, the example uses a short text, but the function accepts any source.
def chunk(text: str, size: int = 400, overlap: int = 50) -> list[str]:
"""Quebra texto em chunks com overlap, sem cortar palavra."""
words = text.split()
out, i = [], 0
while i < len(words):
out.append(" ".join(words[i:i + size]))
i += size - overlap
return out
def embed(texts: list[str]) -> list[list[float]]:
resp = oai.embeddings.create(model=EMBED_MODEL, input=texts)
return [d.embedding for d in resp.data]
def ingest(text: str, source: str):
pieces = chunk(text)
vectors = embed(pieces)
points = [
PointStruct(id=i, vector=v, payload={"text": pieces[i], "source": source})
for i, v in enumerate(vectors)
]
qdrant.upsert(collection_name=COLLECTION, points=points)
print(f"indexado: {len(pieces)} chunks de {source}")
Index an example document. It can be an internal manual, a FAQ, a meeting transcript. The agent will use this as memory.
doc = """
Política de reembolso da Empresa X. Pedidos podem ser reembolsados em até 30 dias
após a compra. O reembolso é processado em até 7 dias úteis no método de pagamento
original. Produtos digitais não são reembolsáveis após o download.
Para solicitar, abra ticket em suporte@empresax.com com o número do pedido.
"""
ingest(doc, source="politica-reembolso-v1")
4. The search tool
The agent does not query the vector DB directly. It calls a tool and your code executes it. That decoupling is what lets you swap the backend (Qdrant for pgvector, for example) without touching the agent.
def search_kb(query: str, k: int = 3) -> list[dict]:
"""Busca k chunks mais relevantes para a query."""
qvec = embed([query])[0]
hits = qdrant.search(
collection_name=COLLECTION,
query_vector=qvec,
limit=k,
)
return [
{"text": h.payload["text"], "source": h.payload["source"], "score": h.score}
for h in hits
]
Declare the tool in the format the model understands (OpenAI tool calling schema).
TOOLS = [{
"type": "function",
"function": {
"name": "search_kb",
"description": "Busca trechos relevantes na base de conhecimento da empresa.",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Pergunta ou termo a buscar."},
"k": {"type": "integer", "description": "Número de trechos.", "default": 3}
},
"required": ["query"]
}
}
}]
5. The agent loop
Here is the heart. It receives the question, calls the model, and if the model asks for a tool the code executes it and returns the result, and it repeats until the model answers directly.
SYSTEM = """Você é um agente de suporte da Empresa X.
Quando o usuário fizer uma pergunta, SEMPRE busque na base de conhecimento
com search_kb antes de responder. Cite a fonte no final."""
def run_agent(user_msg: str, max_steps: int = 5) -> str:
messages = [
{"role": "system", "content": SYSTEM},
{"role": "user", "content": user_msg},
]
for _ in range(max_steps):
resp = oai.chat.completions.create(
model=CHAT_MODEL,
messages=messages,
tools=TOOLS,
tool_choice="auto",
)
msg = resp.choices[0].message
messages.append(msg)
# se não pediu tool, é resposta final
if not msg.tool_calls:
return msg.content
# executa cada tool pedida
for call in msg.tool_calls:
args = json.loads(call.function.arguments)
if call.function.name == "search_kb":
result = search_kb(**args)
else:
result = {"error": "tool desconhecida"}
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": json.dumps(result, ensure_ascii=False),
})
return "limite de passos atingido"
6. Running it
if __name__ == "__main__":
print(run_agent("Quantos dias eu tenho pra pedir reembolso?"))
print("---")
print(run_agent("Posso devolver produto digital?"))
Expected output (the LLM varies, but the content is grounded):
Você tem até 30 dias após a compra para solicitar reembolso.
Para iniciar, abra um ticket em suporte@empresax.com com o número do pedido.
Fonte: politica-reembolso-v1.
---
Não. Produtos digitais não são reembolsáveis após o download.
Fonte: politica-reembolso-v1.
In two calls the agent did: embedding of the question, search in the vector DB, reading of the most relevant excerpt, generation of a grounded answer and citation of the source. Without the document content in the original prompt.
7. What is missing for production
The agent above is didactic. To take it to production the stack needs to grow on three fronts. Robustness: retry with backoff on API calls, timeout per tool, circuit breaker, sandbox on execution tools. Observability: structured logging of input, retrieved context and answer, traces with OpenTelemetry, p95 latency metrics and cost per turn. Quality: a golden set of 50+ queries with reference answers, a reranker (Cohere Rerank or BGE) between the vector DB and the LLM, continuous evaluation in CI.
Other things that need attention: ACL per chunk (who can see what), PII redaction in the log, prompt injection detection in the input, index versioning when you change the embedding model, and embedding caching for repeated queries.
But that is the skeleton. The agent is a loop: it retrieves, reasons, acts, observes, repeats. Everything that grows in production is robustness around that loop, not a replacement for it. Start small, with this example running, and add complexity when the concrete problem demands it, not before.