Leíste sobre RAG, vector database y tool calling y quieres ver código corriendo. Este post lo entrega. No es arquitectura abstracta ni una diapositiva de keynote, es un agente funcional que recupera contexto de una base de conocimiento y responde con fundamento, escrito en Python, con Qdrant y la API de OpenAI.
Al final, tienes un archivo único que corre local, indexa un documento, recibe una pregunta en la terminal, busca contexto relevante vía embedding, arma el prompt, llama al modelo con tool calling y devuelve la respuesta con la fuente citada. Cerca de 120 líneas, sin framework. La intención es didáctica: después lo cambias por LangGraph, LlamaIndex o un framework propio sabiendo lo que hay debajo.
1. Stack e instalación
Tres piezas. El OpenAI SDK para embedding y LLM, Qdrant en Docker como vector database, y Python 3.11+. Instala las dependencias y levanta 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-...
Crea un archivo agent.py. Estructura: setup, ingesta, tool, loop. Vamos por etapas.
2. Setup y 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 ingesta
Aquí está lo que separa una demo de algo serio: chunking respetando la frontera semántica y enriquecimiento con metadato. Para simplificar, el ejemplo usa un texto corto, pero la función acepta cualquier fuente.
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}")
Indexa un documento de ejemplo. Puede ser un manual interno, un FAQ, la transcripción de una reunión. El agente lo va a usar como memoria.
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. La tool de búsqueda
El agente no consulta el vector DB directamente. Llama a una tool y tu código la ejecuta. Ese desacoplamiento es lo que permite cambiar el backend (Qdrant por pgvector, por ejemplo) sin tocar el 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
]
Declara la tool en el formato que el modelo entiende (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. El loop del agente
Aquí está el corazón. Recibe la pregunta, llama al modelo, si él pide una tool el código la ejecuta y devuelve el resultado, y repite hasta que responda directo.
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. Ejecutándolo
if __name__ == "__main__":
print(run_agent("Quantos dias eu tenho pra pedir reembolso?"))
print("---")
print(run_agent("Posso devolver produto digital?"))
Salida esperada (el LLM varía, pero el contenido es 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.
En dos llamadas el agente hizo: embedding de la pregunta, búsqueda en el vector DB, lectura del fragmento más relevante, generación de respuesta fundamentada y cita de la fuente. Sin el contenido del documento en el prompt original.
7. Lo que falta para producción
El agente de arriba es didáctico. Para subir a producción el stack necesita crecer en tres frentes. Robustez: retry con backoff en las llamadas de API, timeout por tool, circuit breaker, sandbox en tools de ejecución. Observabilidad: log estructurado de input, contexto recuperado y respuesta, traces con OpenTelemetry, métricas de latencia p95 y costo por turn. Calidad: golden set de 50+ queries con respuesta de referencia, reranker (Cohere Rerank o BGE) entre el vector DB y el LLM, evaluación continua en CI.
Otras cosas que necesitan atención: ACL por chunk (quién puede ver qué), redaction de PII en el log, detección de prompt injection en el input, versionado del índice cuando cambies el modelo de embedding, y cache de embedding para queries repetidas.
Pero el esqueleto es ese. El agente es un loop: recupera, razona, actúa, observa, repite. Todo lo que crece en producción es robustez alrededor de ese loop, no su reemplazo. Empieza pequeño, con este ejemplo corriendo, y agrega complejidad cuando el problema concreto lo pida, no antes.