Você leu sobre RAG, vector database e tool calling e quer ver código rodando. Este post entrega. Não é arquitetura abstrata nem slide de keynote, é um agente funcional que recupera contexto de uma base de conhecimento e responde com fundamento, escrito em Python, com Qdrant e a API da OpenAI.
No fim, você tem um arquivo único que roda local, indexa um documento, recebe pergunta no terminal, busca contexto relevante via embedding, monta o prompt, chama o modelo com tool calling e devolve a resposta com fonte citada. Cerca de 120 linhas, sem framework. A intenção é didática: depois você troca por LangGraph, LlamaIndex ou framework próprio sabendo o que está embaixo.
1. Stack e instalação
Três peças. OpenAI SDK para embedding e LLM, Qdrant em Docker como vector database, e Python 3.11+. Instale dependências e suba o Qdrant local.
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-...
Crie um arquivo agent.py. Estrutura: setup, ingestão, tool, loop. Vamos por etapas.
2. Setup e cliente Qdrant
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. Pipeline de ingestão
Aqui está o que separa demo de coisa séria: chunking respeitando fronteira semântica e enriquecimento com metadado. Pra simplificar, o exemplo usa um texto curto, mas a função aceita qualquer fonte.
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}")
Indexe um documento de exemplo. Pode ser um manual interno, um FAQ, transcrição de reunião. O agente vai usar isso como memória.
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. A tool de busca
O agente não consulta o vector DB diretamente. Ele chama uma tool e o seu código executa. Esse desacoplamento é o que permite trocar o backend (Qdrant para pgvector, por exemplo) sem mexer no agente.
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 a tool no formato que o modelo entende (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. O loop do agente
Aqui está o coração. Recebe a pergunta, chama o modelo, se ele pedir uma tool o código executa e devolve resultado, e repete até ele responder direto.
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. Rodando
if __name__ == "__main__":
print(run_agent("Quantos dias eu tenho pra pedir reembolso?"))
print("---")
print(run_agent("Posso devolver produto digital?"))
Saída esperada (o LLM varia, mas o conteúdo é fundamentado):
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.
Em duas chamadas o agente fez: embedding da pergunta, busca no vector DB, leitura do trecho mais relevante, geração de resposta fundamentada e citação da fonte. Sem o conteúdo do documento no prompt original.
7. O que falta pra produção
O agente acima é didático. Pra subir em produção a stack precisa crescer em três frentes. Robustez: retry com backoff em chamadas de API, timeout por tool, circuit breaker, sandbox em tools de execução. Observabilidade: log estruturado de input, contexto recuperado e resposta, traces com OpenTelemetry, métricas de latência p95 e custo por turn. Qualidade: golden set de 50+ queries com gabarito, reranker (Cohere Rerank ou BGE) entre o vector DB e o LLM, avaliação contínua no CI.
Outras coisas que precisam atenção: ACL por chunk (quem pode ver o quê), redaction de PII no log, detecção de prompt injection no input, versionamento do índice quando trocar modelo de embedding, e cache de embedding pra queries repetidas.
Mas o esqueleto é esse. O agente é um loop: recupera, raciocina, age, observa, repete. Tudo o que cresce em produção é robustez ao redor desse loop, não substituição dele. Comece pequeno, com este exemplo rodando, e adicione complexidade quando o problema concreto pedir, não antes.
