RAG (Retrieval-Augmented Generation): Cómo Crear Chatbots que Conocen tu Empresa

from langchain.document_loaders import (
    PyPDFLoader,
    Docx2txtLoader,
    TextLoader,
    UnstructuredEmailLoader,
    CSVLoader,
    WebBaseLoader
)

# PDF
pdf_loader = PyPDFLoader("manual_politicas.pdf")
docs = pdf_loader.load()

# Word
word_loader = Docx2txtLoader("procedimiento.docx")
docs = word_loader.load()

# Web
web_loader = WebBaseLoader("https://empresa.com/faq")
docs = web_loader.load()

# Email
email_loader = UnstructuredEmailLoader("comunicado.eml")
docs = email_loader.load()

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,        # Tamaño máximo del chunk
    chunk_overlap=200,      # Superposición entre chunks
    length_function=len,
    separators=["\n\n", "\n", " ", ""]  # Prioridad de separación
)

chunks = splitter.split_documents(docs)

from langchain.embeddings import OpenAIEmbeddings
from langchain.embeddings import HuggingFaceEmbeddings

# OpenAI (mejor calidad, costo por uso)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# Costo: $0.02 / 1M tokens

# Hugging Face (gratuito, local)
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)

# Generar embedding
vector = embeddings.embed_query("¿Cuál es la política de vacaciones?")
# Resultado: [0.023, -0.041, 0.089, ...] (1536 dimensiones)

from langchain.vectorstores import Pinecone, Weaviate, Chroma

# Pinecone (Cloud, enterprise)
import pinecone
pinecone.init(api_key="xxx", environment="us-west1-gcp")
vectorstore = Pinecone.from_documents(
    chunks,
    embeddings,
    index_name="knowledge-base"
)

# Chroma (Local, open source)
vectorstore = Chroma.from_documents(
    chunks,
    embeddings,
    persist_directory="./chroma_db"
)

# Weaviate (Cloud o self-hosted)
vectorstore = Weaviate.from_documents(
    chunks,
    embeddings,
    weaviate_url="http://localhost:8080"
)

import pinecone
from langchain.vectorstores import Pinecone

# Inicializar
pinecone.init(
    api_key="your-api-key",
    environment="us-west1-gcp"
)

# Crear índice
pinecone.create_index(
    name="empresa-kb",
    dimension=1536,
    metric="cosine"
)

# Insertar documentos
vectorstore = Pinecone.from_documents(
    documents=chunks,
    embedding=embeddings,
    index_name="empresa-kb",
    namespace="politicas"  # Multi-tenancy
)

# Búsqueda con filtros
results = vectorstore.similarity_search(
    "política de vacaciones",
    k=5,
    filter={"departamento": "RRHH", "año": 2024}
)

from langchain.vectorstores import Chroma

# Crear base de datos local
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",
    collection_name="knowledge_base"
)

# Persistir
vectorstore.persist()

# Cargar existente
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="knowledge_base"
)

# Búsqueda
results = vectorstore.similarity_search_with_score(
    "¿Cómo solicito vacaciones?",
    k=3
)

for doc, score in results:
    print(f"Score: {score:.3f}")
    print(f"Contenido: {doc.page_content[:200]}...")
    print(f"Fuente: {doc.metadata['source']}")

import weaviate
from langchain.vectorstores import Weaviate

# Conectar
client = weaviate.Client(
    url="http://localhost:8080",
    auth_client_secret=weaviate.AuthApiKey(api_key="xxx")
)

# Crear schema
client.schema.create_class({
    "class": "Document",
    "vectorizer": "text2vec-openai",
    "properties": [
        {"name": "content", "dataType": ["text"]},
        {"name": "source", "dataType": ["string"]},
        {"name": "department", "dataType": ["string"]}
    ]
})

# Búsqueda híbrida (vector + BM25)
results = client.query.get(
    "Document", ["content", "source"]
).with_hybrid(
    query="política de vacaciones",
    alpha=0.75  # 0=keyword, 1=vector
).with_limit(5).do()

import os
from langchain.document_loaders import DirectoryLoader, PyPDFLoader

# Cargar todos los PDFs de una carpeta
loader = DirectoryLoader(
    "./documentos/",
    glob="**/*.pdf",
    loader_cls=PyPDFLoader,
    show_progress=True
)

documents = loader.load()
print(f"Documentos cargados: {len(documents)}")

# Agregar metadata
for doc in documents:
    # Extraer departamento del path
    doc.metadata["departamento"] = doc.metadata["source"].split("/")[2]
    doc.metadata["fecha_carga"] = "2024-06-15"

from langchain.text_splitter import RecursiveCharacterTextSplitter

# Configuración óptima para documentos empresariales
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
    separators=[
        "\n\n",      # Párrafos
        "\n",        # Líneas
        ". ",        # Oraciones
        ", ",        # Cláusulas
        " ",         # Palabras
        ""           # Caracteres
    ]
)

chunks = splitter.split_documents(documents)
print(f"Chunks generados: {len(chunks)}")

# Verificar distribución de tamaños
sizes = [len(c.page_content) for c in chunks]
print(f"Tamaño promedio: {sum(sizes)/len(sizes):.0f} caracteres")
print(f"Tamaño mínimo: {min(sizes)}")
print(f"Tamaño máximo: {max(sizes)}")

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma

# Embeddings
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    openai_api_key="sk-xxx"
)

# Crear vector store
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./empresa_kb",
    collection_name="knowledge_base"
)

# Persistir
vectorstore.persist()
print("Base de conocimiento creada exitosamente")

from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

# LLM
llm = ChatOpenAI(
    model="gpt-4",
    temperature=0,
    openai_api_key="sk-xxx"
)

# Prompt personalizado
prompt_template = """Eres un asistente virtual de la empresa que responde
preguntas basándote ÚNICAMENTE en la información proporcionada.

REGLAS:
1. Solo responde con información del contexto proporcionado
2. Si no encuentras la información, di "No encontré información sobre esto"
3. Cita la fuente del documento cuando sea posible
4. Sé conciso pero completo

CONTEXTO:
{context}

PREGUNTA: {question}

RESPUESTA:"""

PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

# Chain RAG
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(
        search_type="similarity",
        search_kwargs={"k": 5}
    ),
    return_source_documents=True,
    chain_type_kwargs={"prompt": PROMPT}
)

# Consultar
result = qa_chain({"query": "¿Cuál es la política de trabajo remoto?"})
print("Respuesta:", result["result"])
print("\nFuentes:")
for doc in result["source_documents"]:
    print(f"- {doc.metadata['source']}")

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="Knowledge Base API")

class Query(BaseModel):
    question: str
    department: str = None

class Response(BaseModel):
    answer: str
    sources: list
    confidence: float

@app.post("/ask", response_model=Response)
async def ask_question(query: Query):
    try:
        # Filtrar por departamento si se especifica
        search_kwargs = {"k": 5}
        if query.department:
            search_kwargs["filter"] = {"departamento": query.department}

        # Crear retriever con filtros
        retriever = vectorstore.as_retriever(
            search_type="similarity",
            search_kwargs=search_kwargs
        )

        # Ejecutar query
        result = qa_chain({"query": query.question})

        # Calcular confianza basada en scores
        docs_with_scores = vectorstore.similarity_search_with_score(
            query.question, k=5
        )
        avg_score = sum(s for _, s in docs_with_scores) / len(docs_with_scores)
        confidence = max(0, min(1, 1 - avg_score))  # Convertir distancia a confianza

        return Response(
            answer=result["result"],
            sources=[d.metadata["source"] for d in result["source_documents"]],
            confidence=confidence
        )

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/index")
async def index_document(file_path: str):
    """Agregar nuevo documento al índice"""
    loader = PyPDFLoader(file_path)
    new_docs = loader.load()
    new_chunks = splitter.split_documents(new_docs)
    vectorstore.add_documents(new_chunks)
    vectorstore.persist()
    return {"status": "indexed", "chunks": len(new_chunks)}

from langchain.retrievers import BM25Retriever, EnsembleRetriever

# Retriever semántico
semantic_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# Retriever BM25 (keywords)
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5

# Ensemble (combinar ambos)
ensemble_retriever = EnsembleRetriever(
    retrievers=[semantic_retriever, bm25_retriever],
    weights=[0.7, 0.3]  # 70% semántico, 30% keywords
)

# Usar
docs = ensemble_retriever.get_relevant_documents("política de viáticos")

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank

# Reranker de Cohere
compressor = CohereRerank(
    cohere_api_key="xxx",
    top_n=3
)

# Retriever con reranking
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10})
)

# Primero busca 10 docs, luego reordena y devuelve top 3
docs = compression_retriever.get_relevant_documents("¿Cómo solicito vacaciones?")

from langchain.retrievers.multi_query import MultiQueryRetriever

# El LLM genera múltiples versiones de la pregunta
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=llm
)

# Ejemplo: "política de vacaciones" se expande a:
# - "¿Cuál es el proceso para solicitar vacaciones?"
# - "¿Cuántos días de vacaciones tengo?"
# - "¿Qué dice el reglamento sobre días libres?"

docs = multi_query_retriever.get_relevant_documents("vacaciones")

from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo

# Definir metadata disponible
metadata_field_info = [
    AttributeInfo(
        name="departamento",
        description="Departamento al que pertenece el documento (RRHH, Legal, Finanzas)",
        type="string"
    ),
    AttributeInfo(
        name="año",
        description="Año de publicación del documento",
        type="integer"
    ),
    AttributeInfo(
        name="tipo",
        description="Tipo de documento (política, procedimiento, manual)",
        type="string"
    )
]

# Self-query retriever
self_query_retriever = SelfQueryRetriever.from_llm(
    llm=llm,
    vectorstore=vectorstore,
    document_contents="Documentos internos de la empresa",
    metadata_field_info=metadata_field_info
)

# La pregunta "políticas de RRHH del 2024" automáticamente filtra:
# departamento="RRHH" AND año=2024
docs = self_query_retriever.get_relevant_documents(
    "políticas de RRHH del 2024"
)

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

# Store para documentos padre
store = InMemoryStore()

# Splitter para chunks pequeños (búsqueda)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)

# Splitter para documentos más grandes (contexto)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter
)

# Agregar documentos
retriever.add_documents(documents)

# Busca en chunks pequeños, devuelve parents grandes
docs = retriever.get_relevant_documents("política de viáticos")

# Configuración específica para HR
hr_prompt = """Eres el asistente virtual de Recursos Humanos de la empresa.
Respondes preguntas sobre políticas, beneficios, procesos y trámites de RRHH.

REGLAS:
- Responde de forma amigable y profesional
- Si es un tema sensible (despidos, acoso), sugiere hablar con RRHH directamente
- Para trámites, indica los pasos y documentos necesarios
- Si no sabes, di "No tengo esa información, te sugiero contactar a rrhh@empresa.com"

CONTEXTO:
{context}

PREGUNTA DEL EMPLEADO: {question}

RESPUESTA:"""

# Ejemplo de interacción
queries = [
    "¿Cuántos días de vacaciones me corresponden?",
    "¿Cómo solicito un permiso por paternidad?",
    "¿Cuál es el proceso para solicitar trabajo remoto?",
    "¿Qué beneficios de salud tenemos?"
]

# Knowledge base: Fichas de producto, precios, comparativas, casos de éxito

sales_prompt = """Eres un asistente para el equipo de ventas.
Ayudas a encontrar información sobre productos, precios y competencia.

REGLAS:
- Proporciona datos específicos cuando estén disponibles
- Para precios, siempre menciona que pueden variar y consultar lista actualizada
- Incluye diferenciadores vs competencia cuando sea relevante
- Sugiere casos de éxito similares al prospecto

CONTEXTO:
{context}

PREGUNTA DEL VENDEDOR: {question}

RESPUESTA:"""

# Casos de uso
queries = [
    "¿Qué diferencia nuestro producto del competidor X?",
    "Dame los argumentos para justificar el precio premium",
    "¿Tenemos casos de éxito en el sector retail?",
    "¿Cuáles son las objeciones comunes y cómo responderlas?"
]

# Knowledge base: Contratos, normativas, políticas de compliance

legal_prompt = """Eres un asistente para el equipo legal.
Ayudas a encontrar información en contratos, normativas y políticas.

REGLAS CRÍTICAS:
- SIEMPRE incluye disclaimer: "Esta información es referencial, consulte con el equipo legal para decisiones"
- Cita el documento y sección específica
- No des opiniones legales, solo información del contexto
- Para temas de alto riesgo, sugiere revisión de abogado

CONTEXTO:
{context}

PREGUNTA: {question}

RESPUESTA:"""

# Knowledge base: Documentación técnica, manuales, troubleshooting

tech_prompt = """Eres un asistente técnico que ayuda con documentación.

REGLAS:
- Proporciona pasos específicos cuando sea posible
- Incluye código de ejemplo si está disponible
- Para errores, sugiere troubleshooting paso a paso
- Si el problema parece complejo, sugiere escalar a soporte

CONTEXTO:
{context}

PREGUNTA TÉCNICA: {question}

RESPUESTA:"""

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall
)

# Dataset de evaluación
eval_dataset = [
    {
        "question": "¿Cuál es la política de trabajo remoto?",
        "answer": "...",  # Respuesta del sistema
        "contexts": [...],  # Documentos recuperados
        "ground_truth": "..."  # Respuesta correcta esperada
    },
    # ... más ejemplos
]

# Evaluar
results = evaluate(
    eval_dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall
    ]
)

print(results)
# faithfulness: 0.95
# answer_relevancy: 0.88
# context_precision: 0.82
# context_recall: 0.91